64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import logging
|
|
from typing import Generator
|
|
from contextlib import contextmanager
|
|
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
from sqlalchemy import create_engine, text
|
|
|
|
from app.database.exceptions import DatabaseError
|
|
|
|
|
|
class DatabaseManager():
|
|
|
|
_instance: 'DatabaseManager' = None
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if cls._instance is None:
|
|
cls._instance = super().__new__(cls)
|
|
return cls._instance
|
|
|
|
|
|
def __init__(self, mysql_user: str, mysql_password: str, mysql_host: str, mysql_port: str, mysql_db_name: str) -> None:
|
|
self.logger = logging.getLogger(__name__)
|
|
self.logger.info("Reading database config")
|
|
self.engine = create_engine('mysql+mysqlconnector://%s:%s@%s:%s/%s' % (
|
|
mysql_user,
|
|
mysql_password,
|
|
mysql_host,
|
|
mysql_port,
|
|
mysql_db_name),
|
|
pool_pre_ping=True)
|
|
self.test_connection()
|
|
self.Session = sessionmaker(bind=self.engine) # NOSONAR
|
|
|
|
def cleanup(self) -> None:
|
|
self.logger.debug("Closing connection")
|
|
self.engine.dispose()
|
|
|
|
def test_connection(self):
|
|
self.logger.debug("Testing database connection")
|
|
try:
|
|
with self.engine.connect() as connection:
|
|
connection.execute(text("select 1"))
|
|
self.logger.debug("Database connection successful")
|
|
except DatabaseError as e:
|
|
self.logger.critical("Database connection failed: %s", e)
|
|
raise DatabaseError("Database connection failed", DatabaseError.CONNECTION_ERROR) from e
|
|
|
|
|
|
@classmethod
|
|
@contextmanager
|
|
def get_session(cls) -> Generator[Session, None, None]:
|
|
session = cls._instance.Session()
|
|
try:
|
|
yield session
|
|
except Exception as e:
|
|
session.rollback()
|
|
cls._instance.logger.error("Transaction failed: %s", e)
|
|
raise
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
__all__ = ["DatabaseManager"]
|