kontor/kontor-scripts/api.py

from dataclasses import dataclass
from enum import Enum, auto
from logging import Logger
from pathlib import Path
from typing import Dict, List, Optional
from uuid import UUID

from platformdirs import PlatformDirs
import requests
import yaml


MAPPING: Dict[str, str] = {
    "sport": "api/tysc/sports",
    "player": "api/tysc/players",
    "team": "api/tysc/teams",
    "field_position": "api/tysc/positions",
    "rooster": "api/tysc/roosters",
    "vendor": "api/tysc/vendors",
    "card_set": "api/tysc/cardsets",
    "card": "api/tysc/cards",
    "artist": "api/comics/artists",
    "publisher": "api/comics/publishers",
    "worktype": "api/comics/worktypes",
    "comic": "api/comics/comics",
    "volume": "api/comics/volumes",
    "story_arc": "api/comics/storyarcs",
    "issue": "api/comics/issues",
    "comic_work": "api/comics/comicworks",
    "issue_work": "api/comics/issueworks",
    "article": "api/bookshelf/articles",
    "bookshelf_publisher": "api/bookshelf/publishers",
    "book": "api/bookshelf/books",
    "author": "api/bookshelf/authors",
    "article_author": "api/bookshelf/articleauthors",
    "book_author": "api/bookshelf/bookauthors",
    "media_article": "api/media/articles",
    "media_video": "api/media/videos",
    "media_lofi": "api/media/lofi",
    "media_file": "api/media/files",
    "media_actor": "api/media/actors",
    "media_actor_file": "api/media/actorfiles",
    "profile": "api/user/profiles",
    "permission": "api/user/permissions",
    "assignment": "api/user/assignments",
    "token": "api/user/tokens",
    "mail_account": "api/admin/mailaccounts",
}


class OptionType(Enum):
    """
    OptionType defines the type of param for REST API call.
    The type PARAM indicates a query parameter.
    The type ID indicates the option is an idntifier as part of the path.
    """

    PARAM = auto()
    ID = auto()
    URL = auto()


class Option:
    """
    Option is an utility class to simplify options for the REST API call.
    The type defines how to handle the value.
    """

    def __init__(self, option_type: OptionType, value: str) -> None:
        self.type: Optional[OptionType] = option_type
        self.value: Optional[str] = value

    def __str__(self) -> str:
        if self.type is OptionType.PARAM:
            return f"?{self.value}"
        else:
            return f"/{self.value}"


class CredentialsValidationException(Exception):
    """
    Raised when login failed or token is outdated.
    """


class EndPointNotAvailableException(Exception):
    """
    Raised when calling an not existing endpoint.
    """


@dataclass
class Login:
    """
    Dataclass to store login information.
    """

    email: str
    password: str


@dataclass
class Server:
    """
    Dataclass to represent a Kontor-API instance.
    """

    name: str
    url: str
    token: str
    token_type: str
    email: str
    password: str
    timeout: int

    def login(self, log: Logger, refresh_token: bool = False):
        """
        get token from server by calling login endpoint.
        """
        log.debug("call login and retrieve token")
        if not self.token:
            self.__get_token__(log=log)
        if refresh_token:
            self.__get_token__(log=log)

    def __get_token__(self, log: Logger):
        log.info("Call login first")
        login_url = f"{self.url}/login"
        login_data = {}
        login_data["email"] = self.email
        login_data["password"] = self.password
        response = requests.post(login_url, json=login_data, timeout=self.timeout)
        status = response.status_code
        log.info(f"Status: {status}")
        if status != 200:
            log.fatal("authentication failed")
            return
        data = response.json()
        log.debug(f"got data: {data}")
        token = data["access_token"]
        token_type = data["token_type"]
        self.token = str(token)
        self.token_type = str(token_type)

    def request(self, log: Logger, table: str, param: Optional[Option] = None):
        """
        Requests data from Kontor-API instance by given table and optional parameters.
        """
        if not param:
            url: str = f"{self.url}/{MAPPING[table]}"
        else:
            url: str = f"{self.url}/{MAPPING[table]}{str(param)}"
        headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
        response = requests.get(url, headers=headers, timeout=self.timeout)
        log.debug(f"Status: {response.status_code}")
        if response.status_code == 401:
            self.login(log, refresh_token=True)
            headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
            response = requests.get(url, headers=headers, timeout=self.timeout)
        if response.status_code == 404:
            raise EndPointNotAvailableException
        data = response.json()
        return data

    def update(self, log: Logger, table: str, item_id: UUID, file_info: dict):
        """
        Updates data to the Kontor-API instance.
        """
        url: str = f"{self.url}/{MAPPING[table]}/{item_id}"
        headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
        update = requests.put(
            url, headers=headers, json=file_info, timeout=self.timeout
        )
        log.info(f"Status: {update.status_code}")
        if update.status_code == 404:
            raise EndPointNotAvailableException
        data = update.json()
        return data

    def create(self, log: Logger, table: str, new_item: dict):
        """
        Create item in Kontor-API instance.
        """
        url: str = f"{self.url}/{MAPPING[table]}"
        headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
        create = requests.post(
            url, headers=headers, json=new_item, timeout=self.timeout
        )
        log.info(f"Status: {create.status_code}")
        match create.status_code:
            case 404:
                log.info("Endpoint %s not available", url)
                raise EndPointNotAvailableException
            case 405:
                log.info("POST request %s returned 405", url)
            case 409:
                log.fatal("Create Exception %s", create.json())
            case _:
                log.info("")
        data = create.json()
        return data

    def delete(self, log: Logger, table: str, item_id: str):
        """
        Delete item in Kontor-API instance.
        """
        url: str = f"{self.url}/{MAPPING[table]}/{item_id}"
        headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
        delete = requests.delete(url, headers=headers, timeout=self.timeout)
        log.info(f"Status: {delete.status_code}")


@dataclass
class ApiConfig:
    """
    Dataclass to define required contents of configuration file.
    """

    server: List[Server]

    def get_server(self, server_name: str) -> Optional[Server]:
        """
        Return server instance by given name or None.
        """
        found_server = None
        for server in self.server:
            if server.name == server_name:
                found_server = server
        return found_server


def get_api_config(log: Logger, config: str) -> ApiConfig:
    """
    Load configuration from file.
    """
    dirs = PlatformDirs(config)
    api_config = Path(dirs.user_config_dir, "api.yaml")
    with open(api_config, "rt", encoding="utf-8") as f:
        api_data = yaml.safe_load(f.read())
        servers = [Server(**server) for server in api_data["server"]]
        api_config_data = ApiConfig(server=servers)
        log.debug(api_config_data)
        if not api_data:
            log.fatal("API configuration is missing")
            return api_config_data
    for server in api_config_data.server:
        server.login(log)
    with open(api_config, "w", encoding="utf-8") as f:
        yaml.dump(api_data, f)
    return api_config_data