kontor/kontor-scripts/api.py

from dataclasses import dataclass
import logging
import logging.config
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",
    "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_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 EndPointNotAvailableException(Exception):
    """
    Raised when calling an not existing endpoint.
    """
    
    pass

@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
    timeout: int

    def login(self, login: Login, log: Logger):
        """
        get token from server by calling login endpoint.
        """
        if not self.token:
            log.info("Call login first")
            login_url = f"{self.url}/login"
            login_data = {}
            login_data["email"] = login.email
            login_data["password"] = login.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[str] = None):
        if not param:
            url: str = f"{self.url}/{MAPPING[table]}"
        else:
            url: str = f"{self.url}/{MAPPING[table]}?{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==404:
            raise EndPointNotAvailableException
        data = response.json()
        return data

    def update(self, log: Logger, table: str, item_id: UUID, file_info: dict):
        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}")
        return update


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

    login: Login
    server: List[Server]
    
    def get_server(self, server_name: str) -> Optional[Server]:
        """
        """
        found_server = None
        for server in self.server:
            if server.name == server_name:
                found_server = server
        return found_server
        


def get_logger(level, config: str):
    """
    get Logger according to given log level by verbosity.
    """
    dirs = PlatformDirs(config)
    logging_config = Path(dirs.user_config_dir, "logging-config.yaml")
    log_config = None
    with open(logging_config, "rt", encoding="utf-8") as f:
        log_config = yaml.safe_load(f.read())
    logging.config.dictConfig(log_config)
    logger = logging.getLogger("development")
    if level is not None:
        match level:
            case 0:
                logger.setLevel(logging.CRITICAL)
            case 1:
                logger.setLevel(logging.INFO)
            case 2:
                logger.setLevel(logging.DEBUG)
            case _:
                logger.setLevel(logging.INFO)
    return logger


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