from dataclasses import dataclass from enum import Enum, auto 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", "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_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): PARAM = auto() ID = auto() class Option: 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 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[Option] = None): 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 == 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: logging.getLogger("requests").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) 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