config di lifespan

api/infrastructure/dependencies/adapters.py

@@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import (
 from api.application.abstractions import UnitOfWork
 from api.infrastructure.dependencies.stub import Stub
 from api.infrastructure.persistence.uow import SqlAlchemyUnitOfWork
+from api.infrastructure.settings import Settings
 
 
 def new_unit_of_work(
@@ -20,15 +21,16 @@ def new_unit_of_work(
     return SqlAlchemyUnitOfWork(session)
 
 
-def create_engine() -> AsyncEngine:
-    return create_async_engine("postgresql+asyncpg://postgresql+asyncpg//demo_user:user_pass@db:5432/serviceman_db")
+def create_engine(
+    settings: Annotated[Settings, Depends(Stub(Settings))],
+) -> AsyncEngine:
+    return create_async_engine(settings.db.db_url)
 
 
 def create_session_maker(
     engine: Annotated[AsyncEngine, Depends(Stub(AsyncEngine))],
 ) -> async_sessionmaker[AsyncSession]:
     maker = async_sessionmaker(engine, expire_on_commit=False)
-    print("session_maker id:", id(maker))
     return maker
 
 

api/infrastructure/dependencies/configs.py

@@ -0,0 +1,30 @@
+import os
+from functools import lru_cache
+
+import yaml  # type: ignore
+
+from api.infrastructure.persistence.db_setings import DBSettings
+from api.infrastructure.settings import Settings
+
+
+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
+
+
+@lru_cache
+def app_settings() -> Settings:
+    config_data = yaml_loader(
+        file=os.getenv("CONFIG_PATH", "./config/api_config.yml"),
+    )
+
+    return Settings(
+        db=DBSettings(
+            pg_user=config_data["db"]["user"],
+            pg_pass=config_data["db"]["password"],
+            pg_host=config_data["db"]["host"],
+            pg_port=int(config_data["db"]["port"]),
+            pg_db=config_data["db"]["database"],
+        ),
+    )

api/infrastructure/dependencies/usecases.py

@@ -5,10 +5,11 @@ from fastapi import Depends
 from api.application.abstractions.uow import UnitOfWork
 from api.application.usecase.create_user import CreateUser
 from api.domain.user.repository import UserRepository
+from api.infrastructure.dependencies.stub import Stub
 
 
 def provide_create_user(
-    user_repository: Annotated[UserRepository, Depends()],
+    user_repository: Annotated[UserRepository, Depends(Stub(UserRepository))],
     uow: Annotated[UnitOfWork, Depends()],
 ) -> CreateUser:
     return CreateUser(uow=uow, user_repository=user_repository)