src/di/providers.py - master - urusai/code_sample

src/di/providers.py
59 строк2 KB
nick
linters
30 сен 2024, 16:13
30 сен 2024, 16:1309a0f4a
import logging

from fastapi import Depends
from alembic_api import AlembicClient, AlembicServer

from di.factory import ConfigFactory, EngineFactory, SessionFactory
from db.repositories.base import BaseSession
from settings.db.settings import SettingsPostgres
from settings.api.settings import SettingsApplication
from settings.redis.settings import SettingsRedis
from services.http_client.client import HTTPClient

logger = logging.getLogger(__name__)


class FastAPIDependenciesProvider:
    def __init__(self, config: ConfigFactory):
        self.config = config

        logger.info("Creating instance Engine(mode=async)")
        self.async_engine_factory = EngineFactory(config=config.postgres, mode="async")

        logger.info("Creating instance Engine(mode=sync)")
        self.sync_e_factory = EngineFactory(config=config.postgres, mode="sync")
        logger.info(
            "Creating "
            f"async_sessionmaker("
            f"dbms={self.async_engine_factory.engine.name}, "
            f"driver={self.async_engine_factory.engine.driver}"
            ")"
        )
        self.async_session_factory = SessionFactory(
            engine=self.async_engine_factory.engine
        )

    def get_async_engine_factory(self) -> EngineFactory:
        return self.async_engine_factory

    def get_session_factory(self) -> SessionFactory:
        return self.async_session_factory

    def get_settings_application(self) -> SettingsApplication:
        return self.config.application

    def get_settings_postgres(self) -> SettingsPostgres:
        return self.config.postgres

    def get_settings_redis(self) -> SettingsRedis:
        return self.config.redis

    @staticmethod
    def get_http_client():
        return HTTPClient()

    @staticmethod
    def get_db_session(
        session: SessionFactory = Depends(SessionFactoryMarker),
    ):
        return BaseSession(db_session=session.session_factory)