uow and di basic implementation

api/app.py

@@ -1,12 +1,11 @@
 from fastapi import FastAPI
 
-from api.di import Container
 from api.router.user import router as user_router
 
 
 def create_app() -> FastAPI:
     app = FastAPI()
-    app.container = Container()
+
     app.include_router(user_router)
     return app
 

api/config.py

@@ -0,0 +1,37 @@
+import os
+
+import yaml  # type: ignore
+from pydantic_settings import BaseSettings
+
+with open(os.getenv("CONFIG_PATH", "")) as f:
+    config_data: dict = yaml.safe_load(f)
+
+if os.getenv("INDOCKER"):
+    config_data["db"]["host"] = "db"
+    config_data["db"]["port"] = 5432
+
+
+class DBSettings(BaseSettings):
+    pg_user: str = config_data["db"]["user"]
+    pg_pass: str = config_data["db"]["password"]
+    pg_host: str = config_data["db"]["host"]
+    pg_port: int = config_data["db"]["port"]
+    pg_db: str = config_data["db"]["database"]
+
+    @property
+    def get_db_url(self) -> str:
+        return "postgresql+asyncpg://{}:{}@{}:{}/{}".format(
+            self.pg_user,
+            self.pg_pass,
+            self.pg_host,
+            self.pg_port,
+            self.pg_db,
+        )
+
+
+settings = DBSettings()
+
+
+def get_settings():
+    print(id(settings))
+    return settings

api/di.py

@@ -1,58 +1,28 @@
-import os
-
-import yaml  # type: ignore
-from dependency_injector import containers, providers
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
+from api.config import get_settings
 from api.service.user import UserService
 from api.uow.uow_base import UnitOfWork
 
-
+async_engine = create_async_engine(
-class Container(containers.DeclarativeContainer):
+    url=get_settings().get_db_url,
-    wiring_config = containers.WiringConfiguration(modules=["api.router.user"])
-
-    if not os.getenv("CONFIG_PATH"):
-        raise ValueError('Please set "CONFIG_PATH" variable in your environment')
-
-    with open(os.getenv("CONFIG_PATH", "")) as f:
-        config_data = yaml.safe_load(f)
-
-    config = providers.Configuration()
-
-    if os.getenv("INDOCKER"):
-        config_data["db"]["host"] = "db"
-        config_data["db"]["port"] = 5432
-
-    async_engine = providers.Factory(
-        create_async_engine,
-        "postgresql+asyncpg://{}:{}@{}:{}/{}".format(
-            config_data["db"]["user"],
-            config_data["db"]["password"],
-            config_data["db"]["host"],
-            config_data["db"]["port"],
-            config_data["db"]["database"],
-        ),
     echo=True,
-    )
+)
 
-    async_session_factory = providers.Factory(
+async_session_factory = async_sessionmaker(
-        async_sessionmaker,
     async_engine,
     class_=AsyncSession,
     expire_on_commit=False,
-    )
+)
 
-    uow = providers.Factory(
+uow = UnitOfWork(
-        UnitOfWork,
     session_factory=async_session_factory,
-    )
+)
-    #
+
-    # user_repository = providers.Factory(
+user_service = UserService(
-    #     UserRepository,
-    #     uow=uow,
-    # )
-    #
-    user_service = providers.Factory(
-        UserService,
     uow=uow,
-    )
+)
+
+
+def get_user_service():
+    return user_service

api/router/user.py

@@ -1,7 +1,6 @@
-from dependency_injector.wiring import Provide, inject
 from fastapi import APIRouter, Depends
 
-from api.di import Container
+from api.di import get_user_service
 from api.schemas.user_schema import UserSchema
 from api.service.user import UserService
 
@@ -9,9 +8,8 @@ router = APIRouter()
 
 
 @router.get("/users", response_model=list[UserSchema])
-@inject
 async def get_user_list(
-    user_service: UserService = Depends(Provide[Container.user_service]),
+    user_service: UserService = Depends(get_user_service),
 ) -> list[UserSchema]:
     return await user_service.get_all_users()
 

api/service/user.py

@@ -1,8 +1,8 @@
-from api.uow.uow_base import IUnitOfWork
+from api.uow.uow_base import UnitOfWork
 
 
 class UserService:
-    def __init__(self, uow: IUnitOfWork):
+    def __init__(self, uow: UnitOfWork):
         self.uow = uow
 
     async def get_all_users(self):

api/uow/repository.py

@@ -1,4 +1,3 @@
-from abc import ABC, abstractmethod
 from typing import Generic, TypeVar
 
 from sqlalchemy import insert, select
@@ -9,17 +8,7 @@ import api.models as models
 ModelType = TypeVar("ModelType", bound=models.Base)
 
 
-class AbstractRepository(ABC):
+class SQLAlchemyRepository(Generic[ModelType]):
-    @abstractmethod
-    async def add_one(self, data: dict):
-        raise NotImplementedError()
-
-    @abstractmethod
-    async def find_all(self):
-        raise NotImplementedError()
-
-
-class SQLAlchemyRepository(AbstractRepository, Generic[ModelType]):
     model: type[ModelType]
 
     def __init__(self, session: AsyncSession):

api/uow/uow_base.py

@@ -1,32 +1,6 @@
-from abc import ABC, abstractmethod
-
 from api.repository.user import UserRepository
 
 
-class IUnitOfWork(ABC):
-    users: type[UserRepository]
-
-    @abstractmethod
-    def __init__(self):
-        ...
-
-    @abstractmethod
-    async def __aenter__(self):
-        ...
-
-    @abstractmethod
-    async def __aexit__(self):
-        ...
-
-    @abstractmethod
-    async def commit(self):
-        ...
-
-    @abstractmethod
-    async def rollback(self):
-        ...
-
-
 class UnitOfWork:
     def __init__(self, session_factory):
         self.session_factory = session_factory

config/api_config.yml

@@ -4,4 +4,3 @@ db:
   database: "serviceman_db"
   user: "demo_user"
   password: "user_pass"
-  echo: true