69 lines
2.0 KiB
Python
69 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 sqlalchemy.exc import DatabaseError as SqlAlchemyDatabaseError
|
|
|
|
from app.database.exceptions import DatabaseError
|
|
from app.models.base_model import Base
|
|
|
|
|
|
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) -> None:
|
|
if hasattr(self, "engine"):
|
|
return
|
|
|
|
self.logger = logging.getLogger(__name__)
|
|
self.logger.info("Initializing Database")
|
|
|
|
self.engine = create_engine('sqlite:///bank.db')
|
|
|
|
self.Session = sessionmaker(bind=self.engine)
|
|
self.create_tables()
|
|
|
|
def create_tables(self):
|
|
self.logger.debug("Creating tables")
|
|
Base.metadata.create_all(self.engine)
|
|
|
|
def cleanup(self) -> None:
|
|
self.engine.dispose()
|
|
|
|
def test_connection(self) -> bool:
|
|
self.logger.debug("Testing database connection")
|
|
try:
|
|
with self.engine.connect() as connection:
|
|
connection.execute(text("select 1"))
|
|
self.logger.debug("Database connection successful")
|
|
return True
|
|
except SqlAlchemyDatabaseError as e:
|
|
self.logger.critical("Database connection failed: %s", e)
|
|
raise DatabaseError("Database connection failed", DatabaseError.CONNECTION_ERROR) from e
|
|
|
|
return False
|
|
|
|
@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"]
|