misc: Use single SQLAlchemy engine and session maker

As recommended by SQLAlchemy [1], this change makes a single
instantiation of the database engine and session maker, instead of one
entity per handler.

It also uses the provided `URL` constructor to better define the
database URL structure.

[1] https://docs.sqlalchemy.org/en/20/core/connections.html#basic-usage
This commit is contained in:
Michael Manganiello
2024-08-21 09:52:35 -03:00
parent 419f3baf4b
commit a85c84a7d4
3 changed files with 15 additions and 10 deletions

View File

@@ -2,7 +2,6 @@ import os
import sys
from pathlib import Path
from typing import Final
from urllib.parse import quote_plus
import pydash
import yaml
@@ -21,6 +20,7 @@ from exceptions.config_exceptions import (
ConfigNotWritableException,
)
from logger.logger import log
from sqlalchemy import URL
from yaml.loader import SafeLoader
ROMM_USER_CONFIG_PATH: Final = f"{ROMM_BASE_PATH}/config"
@@ -76,7 +76,7 @@ class ConfigManager:
sys.exit(5)
@staticmethod
def get_db_engine() -> str:
def get_db_engine() -> URL:
"""Builds the database connection string depending on the defined database in the config.yml file
Returns:
@@ -90,9 +90,13 @@ class ConfigManager:
)
sys.exit(3)
return (
f"mariadb+mariadbconnector://{DB_USER}:%s@{DB_HOST}:{DB_PORT}/{DB_NAME}"
% quote_plus(DB_PASSWD)
return URL.create(
drivername="mariadb+mariadbconnector",
username=DB_USER,
password=DB_PASSWD,
host=DB_HOST,
port=DB_PORT,
database=DB_NAME,
)
# DEPRECATED

View File

@@ -1,6 +1,7 @@
import functools
from fastapi import HTTPException, status
from handler.database.base_handler import sync_session
from logger.logger import log
from sqlalchemy.exc import ProgrammingError
@@ -12,7 +13,7 @@ def begin_session(func):
return func(*args, **kwargs)
try:
with args[0].session.begin() as s:
with sync_session.begin() as s:
kwargs["session"] = s
return func(*args, **kwargs)
except ProgrammingError as exc:

View File

@@ -2,8 +2,8 @@ from config.config_manager import ConfigManager
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
sync_engine = create_engine(ConfigManager.get_db_engine(), pool_pre_ping=True)
sync_session = sessionmaker(bind=sync_engine, expire_on_commit=False)
class DBBaseHandler:
def __init__(self) -> None:
self.engine = create_engine(ConfigManager.get_db_engine(), pool_pre_ping=True)
self.session = sessionmaker(bind=self.engine, expire_on_commit=False)
class DBBaseHandler: ...