78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
import os
|
|
from collections.abc import Iterable
|
|
|
|
import yaml # type: ignore
|
|
from dishka import Container, Provider, Scope, make_container, provide
|
|
from sqlalchemy import Engine, create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
|
|
from flask_demo_api.config import Settings
|
|
from flask_demo_api.protocols.repository import RepositoryGateway
|
|
from flask_demo_api.repository.config import DBSettings
|
|
from flask_demo_api.repository.repository import SqlalchemyRepositoryGateway
|
|
from flask_demo_api.usecase.get_gender import GetGender
|
|
|
|
|
|
def yaml_loader(file: str) -> dict[str, dict[str, str]]:
|
|
with open(file) as f:
|
|
yaml_data: dict = yaml.safe_load(f)
|
|
return yaml_data
|
|
|
|
|
|
class DBSettingsProvider(Provider):
|
|
@provide(scope=Scope.APP)
|
|
def db_settings(self) -> DBSettings:
|
|
config_data = yaml_loader(
|
|
file=os.getenv("CONFIG_PATH", "./config/api_config.yml"),
|
|
)
|
|
return DBSettings(
|
|
pg_user=config_data["db"]["pg_user"],
|
|
pg_pass=config_data["db"]["pg_pass"],
|
|
pg_host=config_data["db"]["pg_host"],
|
|
pg_port=int(config_data["db"]["pg_port"]),
|
|
pg_db=config_data["db"]["pg_db"],
|
|
)
|
|
|
|
|
|
class SettingsProvider(Provider):
|
|
@provide(scope=Scope.APP)
|
|
def main_settings(self, db_settings: DBSettings) -> Settings:
|
|
return Settings(db=db_settings)
|
|
|
|
|
|
class SqlalchemyProvider(Provider):
|
|
@provide(scope=Scope.APP)
|
|
def provide_engine(self, config: Settings) -> Engine:
|
|
return create_engine(config.db.get_db_url)
|
|
|
|
@provide(scope=Scope.APP)
|
|
def provide_sessionmaker(self, engine: Engine) -> sessionmaker[Session]:
|
|
return sessionmaker(bind=engine, expire_on_commit=False, class_=Session)
|
|
|
|
@provide(scope=Scope.REQUEST, provides=Session)
|
|
def provide_session(self, sessionmaker: sessionmaker[Session]) -> Iterable[Session]:
|
|
with sessionmaker() as session:
|
|
yield session
|
|
|
|
|
|
class RepositoryProvider(Provider):
|
|
@provide(scope=Scope.REQUEST)
|
|
def provide_repository(self, session: Session) -> RepositoryGateway:
|
|
return SqlalchemyRepositoryGateway(session=session)
|
|
|
|
|
|
class UseCasesProvider(Provider):
|
|
scope = Scope.REQUEST
|
|
|
|
get = provide(GetGender)
|
|
|
|
|
|
def create_container() -> Container:
|
|
return make_container(
|
|
DBSettingsProvider(),
|
|
SettingsProvider(),
|
|
SqlalchemyProvider(),
|
|
RepositoryProvider(),
|
|
UseCasesProvider(),
|
|
)
|