database_service.py

import logging
from typing import Text

import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.engine.url import make_url

from conf.app import config


class SingletonDecorator:
    def __init__(self, cls):
        self.cls      = cls
        self.instance = None

    def __call__(self, *args, **kwargs):
        if self.instance is None:
            self.instance = self.cls(*args, **kwargs)

        return self.instance


def singleton(cls):
    return SingletonDecorator(cls)


@singleton
class DatabaseService:
    def __init__(self, *, debug_database: bool = None):
        self._logger: logging.Logger             = logging.getLogger(self.__module__)
        self._engine_url: sa.engine.url.URL      = config.database_url
        self._debug_database: bool               = debug_database or config.DEBUG_DATABASE
        self._engine: sa.engine.Engine           = None
        self._session_maker: orm.sessionmaker    = None
        self._session: orm.Session               = None

        self._logger.debug(f'Created {self}')

    def __repr__(self):
        return f'<{self.__class__.__name__} object at 0x{id(self):02x}>'

    def _wrap_session(self, session: orm.Session) -> orm.Session:
        close = session.close

        def close_callback():
            close()

            self._logger.debug(f'Closed {self._session}')
            self._session = None

        session.close = close_callback

        return session

    @property
    def engine_url(self) -> sa.engine.url.URL:
        return self._engine_url

    @property
    def engine(self) -> sa.engine.Engine:
        if self._engine is None:
            self._engine = sa.create_engine(self.engine_url)

            self._logger.debug(f'Created <{self._engine} object at 0x{id(self._engine):02x}>')

            if self._debug_database:
                engine_logger           = logging.getLogger('sqlalchemy.engine')
                engine_logger.propagate = False

            self._logger.debug('Configuring ORM mappers')
            orm.configure_mappers()

        return self._engine

    @property
    def session_maker(self) -> orm.sessionmaker:
        if self._session_maker is None:
            self._session_maker = orm.sessionmaker(autoflush=False, bind=self.engine)

            self._logger.debug(f'Created <{self._session_maker} object at 0x{id(self._session_maker):02x}>')

        return self._session_maker

    @property
    def session(self) -> orm.Session:
        if self._session is None:
            self._session = self._wrap_session(self.session_maker())

            self._logger.debug(f'Created {self._session}')

        return self._session

    @property
    def database_name(self) -> Text:
        return self.engine_url.database