278 lines
8.8 KiB
Python
278 lines
8.8 KiB
Python
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):
|
|
"""
|
|
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}")
|
|
if create.status_code == 404:
|
|
raise EndPointNotAvailableException
|
|
if create.status_code == 409:
|
|
log.fatal("Create Exception %s", create.json())
|
|
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_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)
|
|
logging.getLogger("requests").setLevel(logging.WARNING)
|
|
logging.getLogger("urllib3").setLevel(logging.WARNING)
|
|
case 1:
|
|
logging.getLogger("requests").setLevel(logging.INFO)
|
|
logging.getLogger("urllib3").setLevel(logging.INFO)
|
|
logger.setLevel(logging.INFO)
|
|
case 2:
|
|
logger.setLevel(logging.DEBUG)
|
|
logging.getLogger("requests").setLevel(logging.DEBUG)
|
|
logging.getLogger("urllib3").setLevel(logging.DEBUG)
|
|
case _:
|
|
logger.setLevel(logging.INFO)
|
|
logging.getLogger("requests").setLevel(logging.INFO)
|
|
logging.getLogger("urllib3").setLevel(logging.INFO)
|
|
return logger
|
|
|
|
|
|
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
|