57 lines
2.0 KiB
Python
57 lines
2.0 KiB
Python
"""
|
|
Setup database connections
|
|
"""
|
|
import sqlite3
|
|
import psycopg2
|
|
import logging.config
|
|
from platformdirs import PlatformDirs
|
|
from pathlib import Path
|
|
import yaml
|
|
|
|
|
|
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')
|
|
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
|
|
|