""" Setup database connections """ import sqlite3 import psycopg2 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