This commit is contained in:
@@ -0,0 +1,166 @@
|
||||
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": "",
|
||||
"bookshelf_publisher": "",
|
||||
"book": "",
|
||||
"author": "",
|
||||
"article_author": "",
|
||||
"book_author": "",
|
||||
"media_article": "",
|
||||
"media_video": "api/media/videos",
|
||||
"media_file": "api/media/files",
|
||||
"media_actor": "api/media/actors",
|
||||
"media_actor_file": "api/media/actorfiles",
|
||||
"profile": "",
|
||||
"permission": "",
|
||||
"assignment": "",
|
||||
"token": "",
|
||||
"mail_account": "",
|
||||
}
|
||||
|
||||
|
||||
@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.info(f"Status: {response.status_code}")
|
||||
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_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
|
||||
Reference in New Issue
Block a user