Set up DB connection

This commit is contained in:
Maximilian Keßler 2023-11-22 15:31:36 +01:00
parent 6e562937fb
commit 1f8ad00f92
Signed by: max
GPG key ID: BCC5A619923C0BA5
4 changed files with 161 additions and 0 deletions

52
config.py Normal file
View file

@ -0,0 +1,52 @@
import yaml
import platformdirs
from pathlib import Path
import constants
from log_setup import logger
class DBConfig:
def __init__(self, db_name: str, db_user: str, db_pass: str):
self.db_name = db_name
self.db_user = db_user
self.db_pass = db_pass
def read_db_config() -> DBConfig:
config_dir = Path(platformdirs.user_config_dir(constants.APP_NAME, ensure_exists=True))
config_path = config_dir / constants.DB_CONFIG_FILE_NAME
logger.verbose("DB Configuration read from file {}".format(config_path))
if config_path.exists():
with open(config_path, "r") as f:
config = yaml.safe_load(f)
db_name = config.get('dbname', None)
db_user = config.get('dbuser', None)
db_pass = config.get('dbpass', None)
if db_name is None:
logger.debug("Falling back to default DB name {}".format(constants.DEFAULT_DB_NAME))
db_name = constants.DEFAULT_DB_NAME
if db_user is None:
logger.debug("Falling back to default DB user {}".format(constants.DEFAULT_DB_USER))
db_user = constants.DEFAULT_DB_USER
if db_pass is None:
logger.debug("Falling back to default DB pass {}".format(constants.DEFAULT_DB_PASS))
db_pass = constants.DEFAULT_DB_PASS
logger.debug("Read config values (dbname={}, dbuser={}, dbpass={})".format(db_name, db_user, db_pass))
return DBConfig(db_name, db_user, db_pass)
else:
logger.info(
"No configuration file for database connection found, falling back to default values "
"(dbname={}, dbuser={}, dbpass={}).".format(
constants.DEFAULT_DB_NAME, constants.DEFAULT_DB_USER, constants.DEFAULT_DB_PASS
)
)
logger.info(
"Note: To turn off this message, create a config file at {}".format(config_path)
)
return DBConfig(constants.DEFAULT_DB_NAME, constants.DEFAULT_DB_USER, constants.DEFAULT_DB_PASS)

6
constants.py Normal file
View file

@ -0,0 +1,6 @@
APP_NAME = 'hanabi-league'
DB_CONFIG_FILE_NAME = 'config.yaml'
DEFAULT_DB_NAME = 'hanabi-league'
DEFAULT_DB_USER = 'hanabi-league'
DEFAULT_DB_PASS = 'hanabi-league'

31
database.py Normal file
View file

@ -0,0 +1,31 @@
import psycopg2
from config import read_db_config
from log_setup import logger
class DBConnectionManager:
def __init__(self):
self._conn = None
def connect(self):
config = read_db_config()
logger.debug("Establishing database connection.")
self._conn = psycopg2.connect("dbname='{}' user='{}' password='{}' host='localhost'".format(
config.db_name, config.db_user, config.db_pass
))
logger.debug("Established database connection.")
def get_connection(self):
"""
Get the database connection.
If not already connected, this reads the database config file and connects to the DB.
Otherwise, the already active connection is returned.
"""
if self._conn is None:
self.connect()
return self._conn
# Global instance that will hold our DB connection
conn_manager = DBConnectionManager()

72
log_setup.py Normal file
View file

@ -0,0 +1,72 @@
import logging
import os
import verboselogs
import platformdirs
import constants
class LoggerManager:
def __init__(self, console_level: int = logging.INFO):
self.logger = verboselogs.VerboseLogger(constants.APP_NAME)
self.logger.setLevel(logging.DEBUG)
self.file_formatter = logging.Formatter(
'[%(asctime)s] [PID %(process)s] [%(levelname)7s]: %(message)s'
)
self.info_file_formatter = logging.Formatter(
'[%(asctime)s] [PID %(process)s]: %(message)s'
)
self.console_formatter = logging.Formatter(
'[%(levelname)7s]: %(message)s'
)
self.nothing_formatter = logging.Formatter(
'%(message)s'
)
self.console_handler = logging.StreamHandler()
self.console_handler.setLevel(console_level)
self.console_handler.setFormatter(self.nothing_formatter)
log_dir = platformdirs.user_log_dir(constants.APP_NAME)
os.makedirs(log_dir, exist_ok=True)
self.debug_file_handler = logging.FileHandler(log_dir + "/debug_log.txt")
self.debug_file_handler.setFormatter(self.file_formatter)
self.debug_file_handler.setLevel(logging.DEBUG)
self.verbose_file_handler = logging.FileHandler(log_dir + "/verbose_log.txt")
self.verbose_file_handler.setFormatter(self.file_formatter)
self.verbose_file_handler.setLevel(verboselogs.VERBOSE)
self.info_file_handler = logging.FileHandler(log_dir + "/log.txt")
self.info_file_handler.setFormatter(self.info_file_formatter)
self.info_file_handler.setLevel(logging.INFO)
self.logger.addHandler(self.console_handler)
self.logger.addHandler(self.debug_file_handler)
self.logger.addHandler(self.verbose_file_handler)
self.logger.addHandler(self.info_file_handler)
def set_console_level(self, level: int):
self.console_handler.setLevel(level)
if level == logging.INFO:
self.console_handler.setFormatter(self.nothing_formatter)
else:
self.console_handler.setFormatter(self.console_formatter)
def is_console_level_active(self, level: int):
return level >= self.console_handler.level
def get_logger(self):
return self.logger
# Global logger instance
logger_manager = LoggerManager()
logger = logger_manager.get_logger()