103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
"""
|
|
Setup database connections
|
|
"""
|
|
|
|
import logging.config
|
|
import sqlite3
|
|
from logging import Logger
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
import psycopg2
|
|
import requests
|
|
import yaml
|
|
from platformdirs import PlatformDirs
|
|
|
|
|
|
def get_database_cursors(log, config: str):
|
|
dirs = PlatformDirs(config)
|
|
database_config = Path(dirs.user_config_dir, "database-config.yaml")
|
|
with open(database_config, "rt") as f:
|
|
db_config = yaml.safe_load(f.read())
|
|
sqlite_db = db_config["sqlite"]["file"]
|
|
log.info("using SQLite3 database {}".format(sqlite_db))
|
|
sqlite_conn = sqlite3.connect(
|
|
sqlite_db, detect_types=sqlite3.PARSE_DECLTYPES | sqlite3.PARSE_COLNAMES
|
|
)
|
|
mariadb_conn = None
|
|
postgres_conn = psycopg2.connect(
|
|
f"host={db_config['postgres']['host']} port={db_config['postgres']['port']} user={db_config['postgres']['user']} password={db_config['postgres']['password']} dbname={db_config['postgres']['database']}"
|
|
)
|
|
return sqlite_conn, mariadb_conn, postgres_conn
|
|
|
|
|
|
def create_tables(sqlite_conn, logger, recreate_db, scripts):
|
|
logger.info("create_tables")
|
|
for table_id in scripts:
|
|
create_statement = scripts[table_id]["create"]
|
|
drop_statement = scripts[table_id]["drop"]
|
|
logger.debug(create_statement)
|
|
cursor = sqlite_conn.cursor()
|
|
if recreate_db:
|
|
logger.debug(drop_statement)
|
|
cursor.execute(drop_statement)
|
|
cursor.execute(create_statement)
|
|
|
|
|
|
def get_logger(level, config: str):
|
|
dirs = PlatformDirs(config)
|
|
logging_config = Path(dirs.user_config_dir, "logging-config.yaml")
|
|
log_config = None
|
|
with open(logging_config, "rt") 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) -> Dict[str, Any]:
|
|
api_data: Dict[str, Any] = {}
|
|
token: str | None = None
|
|
host: str | None = None
|
|
port: int = 0
|
|
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())
|
|
if not api_data:
|
|
log.fatal("API configuration is missing")
|
|
return api_data
|
|
host = api_data["host"]
|
|
port = api_data["login_port"]
|
|
if not token:
|
|
log.info("Call login first")
|
|
login_url = f"http://{host}:{port}/login"
|
|
login_data = {}
|
|
login_data["email"] = api_data["email"]
|
|
login_data["password"] = api_data["password"]
|
|
response = requests.post(login_url, json=login_data)
|
|
status = response.status_code
|
|
log.info(f"Status: {status}")
|
|
if status != 200:
|
|
log.fatal("authentication failed")
|
|
return api_data
|
|
data = response.json()
|
|
log.debug(f"got data: {data}")
|
|
token = data["access_token"]
|
|
token_type = data["token_type"]
|
|
api_data["token"] = token
|
|
api_data["token_type"] = token_type
|
|
with open(api_config, "w") as f:
|
|
yaml.dump(api_data, f)
|
|
return api_data
|