Apply di to user module

Author: teamhide

.pre-commit-config.yaml

@@ -32,6 +32,11 @@ repos:
                 "types-redis==4.6.0",
                 "types-ujson==5.9.0",
             ]
+-   repo: https://github.com/pycqa/isort
+    rev: 5.13.2
+    hooks:
+      - id: isort
+        name: isort (python)
 ci:
     autofix_commit_msg: 🎨 [pre-commit.ci] Auto format from pre-commit.com hooks
     autoupdate_commit_msg: ⬆ [pre-commit.ci] pre-commit autoupdate

app/container.py

@@ -0,0 +1,14 @@
+from dependency_injector.containers import DeclarativeContainer, WiringConfiguration
+from dependency_injector.providers import Factory, Singleton
+
+from app.user.adapter.output.persistence.repository_adapter import UserRepositoryAdapter
+from app.user.adapter.output.persistence.sqlalchemy.user import UserSQLAlchemyRepo
+from app.user.application.service.user import UserService
+
+
+class Container(DeclarativeContainer):
+    wiring_config = WiringConfiguration(packages=["app"])
+
+    user_repo = Singleton(UserSQLAlchemyRepo)
+    user_repo_adapter = Factory(UserRepositoryAdapter, user_repo=user_repo)
+    user_service = Factory(UserService, repository=user_repo_adapter)

app/server.py

@@ -1,23 +1,27 @@
-from fastapi import FastAPI, Request, Depends
+from fastapi import Depends, FastAPI, Request
 from fastapi.middleware import Middleware
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
 from app.auth.adapter.input.api import router as auth_router
+from app.container import Container
+from app.user.adapter.input.api import router as user_router
 from core.config import config
 from core.exceptions import CustomException
 from core.fastapi.dependencies import Logging
 from core.fastapi.middlewares import (
-    AuthenticationMiddleware,
     AuthBackend,
-    SQLAlchemyMiddleware,
+    AuthenticationMiddleware,
     ResponseLogMiddleware,
+    SQLAlchemyMiddleware,
 )
-from core.helpers.cache import Cache, RedisBackend, CustomKeyMaker
-from app.user.adapter.input.api import router as user_router
+from core.helpers.cache import Cache, CustomKeyMaker, RedisBackend
 
 
 def init_routers(app_: FastAPI) -> None:
+    container = Container()
+    user_router.container = container
+    auth_router.container = container
     app_.include_router(user_router)
     app_.include_router(auth_router)
 

app/user/adapter/input/api/v1/user.py

@@ -1,15 +1,13 @@
+from dependency_injector.wiring import Provide, inject
 from fastapi import APIRouter, Depends, Query
 
-from core.fastapi.dependencies import (
-    PermissionDependency,
-    IsAdmin,
-)
-from app.user.adapter.input.api.v1.request import LoginRequest, CreateUserRequest
+from app.container import Container
+from app.user.adapter.input.api.v1.request import CreateUserRequest, LoginRequest
 from app.user.adapter.input.api.v1.response import LoginResponse
-from app.user.application.dto import CreateUserResponseDTO
-from app.user.application.dto import GetUserListResponseDTO
-from app.user.application.service.user import UserService
+from app.user.application.dto import CreateUserResponseDTO, GetUserListResponseDTO
 from app.user.domain.command import CreateUserCommand
+from app.user.domain.usecase.user import UserUseCase
+from core.fastapi.dependencies import IsAdmin, PermissionDependency
 
 user_router = APIRouter()
 
@@ -19,27 +17,37 @@
     response_model=list[GetUserListResponseDTO],
     dependencies=[Depends(PermissionDependency([IsAdmin]))],
 )
+@inject
 async def get_user_list(
     limit: int = Query(10, description="Limit"),
     prev: int = Query(None, description="Prev ID"),
+    usecase: UserUseCase = Depends(Provide[Container.user_service]),
 ):
-    return await UserService().get_user_list(limit=limit, prev=prev)
+    return await usecase.get_user_list(limit=limit, prev=prev)
 
 
 @user_router.post(
     "",
     response_model=CreateUserResponseDTO,
 )
-async def create_user(request: CreateUserRequest):
+@inject
+async def create_user(
+    request: CreateUserRequest,
+    usecase: UserUseCase = Depends(Provide[Container.user_service]),
+):
     command = CreateUserCommand(**request.model_dump())
-    await UserService().create_user(command=command)
+    await usecase.create_user(command=command)
     return {"email": request.email, "nickname": request.nickname}
 
 
 @user_router.post(
     "/login",
     response_model=LoginResponse,
 )
-async def login(request: LoginRequest):
-    token = await UserService().login(email=request.email, password=request.password)
+@inject
+async def login(
+    request: LoginRequest,
+    usecase: UserUseCase = Depends(Provide[Container.user_service]),
+):
+    token = await usecase.login(email=request.email, password=request.password)
     return {"token": token.token, "refresh_token": token.refresh_token}

app/user/adapter/output/persistence/repository_adapter.py

@@ -1,11 +1,10 @@
-from app.user.adapter.output.persistence.sqlalchemy.user import UserSQLAlchemyRepo
 from app.user.domain.entity.user import User, UserRead
 from app.user.domain.repository.user import UserRepo
 
 
 class UserRepositoryAdapter:
-    def __init__(self):
-        self.user_repo: UserRepo = UserSQLAlchemyRepo()
+    def __init__(self, *, user_repo: UserRepo):
+        self.user_repo = user_repo
 
     async def get_users(
         self,

app/user/application/service/user.py

@@ -1,23 +1,21 @@
 from app.user.adapter.output.persistence.repository_adapter import UserRepositoryAdapter
-from app.user.application.dto import (
-    LoginResponseDTO,
-)
+from app.user.application.dto import LoginResponseDTO
 from app.user.application.exception import (
-    PasswordDoesNotMatchException,
     DuplicateEmailOrNicknameException,
+    PasswordDoesNotMatchException,
     UserNotFoundException,
 )
 from app.user.domain.command import CreateUserCommand
-from app.user.domain.entity.user import UserRead, User
+from app.user.domain.entity.user import User, UserRead
 from app.user.domain.usecase.user import UserUseCase
 from app.user.domain.vo.location import Location
 from core.db import Transactional
 from core.helpers.token import TokenHelper
 
 
 class UserService(UserUseCase):
-    def __init__(self):
-        self.repository = UserRepositoryAdapter()
+    def __init__(self, *, repository: UserRepositoryAdapter):
+        self.repository = repository
 
     async def get_user_list(
         self,

core/db/mixins/timestamp_mixin.py

@@ -1,17 +1,18 @@
-from sqlalchemy import Column, DateTime, func
-from sqlalchemy.ext.declarative import declared_attr
+from datetime import datetime
 
+from sqlalchemy import DateTime, func
+from sqlalchemy.orm import Mapped, mapped_column
 
-class TimestampMixin:
-    @declared_attr
-    def created_at(cls):
-        return Column(DateTime, default=func.now(), nullable=False)
 
-    @declared_attr
-    def updated_at(cls):
-        return Column(
-            DateTime,
-            default=func.now(),
-            onupdate=func.now(),
-            nullable=False,
-        )
+class TimestampMixin:
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        default=func.now(),
+        nullable=False,
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime,
+        default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )

core/fastapi/dependencies/permission.py

@@ -1,12 +1,14 @@
 from abc import ABC, abstractmethod
 from typing import Type
 
-from fastapi import Request
+from dependency_injector.wiring import Provide, inject
+from fastapi import Depends, Request
 from fastapi.openapi.models import APIKey, APIKeyIn
 from fastapi.security.base import SecurityBase
 from starlette import status
 
-from app.user.application.service.user import UserService
+from app.container import Container
+from app.user.domain.usecase.user import UserUseCase
 from core.exceptions import CustomException
 
 
@@ -34,12 +36,17 @@ async def has_permission(self, request: Request) -> bool:
 class IsAdmin(BasePermission):
     exception = UnauthorizedException
 
-    async def has_permission(self, request: Request) -> bool:
+    @inject
+    async def has_permission(
+        self,
+        request: Request,
+        usecase: UserUseCase = Depends(Provide[Container.user_service]),
+    ) -> bool:
         user_id = request.user.id
         if not user_id:
             return False
 
-        return await UserService().is_admin(user_id=user_id)
+        return await usecase.is_admin(user_id=user_id)
 
 
 class AllowAll(BasePermission):

pyproject.toml

@@ -34,6 +34,9 @@ types-redis = "^4.6.0.20240106"
 [tool.coverage.run]
 omit=["tests/*", "test_*.py", "migrations/*"]
 
+[tool.isort]
+profile="black"
+
 [build-system]
 requires = ["poetry-core>=1.0.0"]
 build-backend = "poetry.core.masonry.api"

tests/app/user/adapter/output/persistence/test_repository_adapter.py

@@ -7,15 +7,15 @@
 from app.user.domain.repository.user import UserRepo
 from tests.support.user_fixture import make_user
 
-repository_adapter = UserRepositoryAdapter()
+user_repo_mock = AsyncMock(spec=UserRepo)
+repository_adapter = UserRepositoryAdapter(user_repo=user_repo_mock)
 
 
 @pytest.mark.asyncio
 async def test_get_users(session: AsyncSession):
     # Given
     limit = 1
     prev = 1
-    user_repo_mock = AsyncMock(spec=UserRepo)
     user = make_user(
         id=1,
         password="password",
@@ -45,7 +45,6 @@ async def test_get_users(session: AsyncSession):
 @pytest.mark.asyncio
 async def test_get_user_by_email_or_nickname(session: AsyncSession):
     # Given
-    user_repo_mock = AsyncMock(spec=UserRepo)
     user = make_user(
         id=1,
         password="password",
@@ -82,7 +81,6 @@ async def test_get_user_by_email_or_nickname(session: AsyncSession):
 @pytest.mark.asyncio
 async def test_get_user_by_id(session: AsyncSession):
     # Given
-    user_repo_mock = AsyncMock(spec=UserRepo)
     user = make_user(
         id=1,
         password="password",
@@ -115,7 +113,6 @@ async def test_get_user_by_id(session: AsyncSession):
 @pytest.mark.asyncio
 async def test_get_user_by_email_and_password(session: AsyncSession):
     # Given
-    user_repo_mock = AsyncMock(spec=UserRepo)
     user = make_user(
         id=1,
         password="password",
@@ -151,7 +148,6 @@ async def test_get_user_by_email_and_password(session: AsyncSession):
 @pytest.mark.asyncio
 async def test_save(session: AsyncSession):
     # Given
-    user_repo_mock = AsyncMock(spec=UserRepo)
     user = make_user(
         id=1,
         password="password",