Merge pull request #20 from Zerohertz/issue#18/feat/transaction

[Feat] Transaction 구성 및 테스트 고도화

Author: Zerohertz

Makefile

@@ -16,7 +16,7 @@ lint:
 test:
 	uv sync --group test
 	export DESCRIPTION=$$(cat README.md) && \
-		uv run pytest \
+		uv run pytest -vv \
 		--cov=app --cov-branch --cov-report=xml \
 		--junitxml=junit.xml -o junit_family=legacy
 

README.md

@@ -34,4 +34,6 @@
 </h4>
 
 - [fastapi/full-stack-fastapi-template/backend](https://github.com/fastapi/full-stack-fastapi-template/tree/master/backend)
+- [teamhide/fastapi-boilerplate](https://github.com/teamhide/fastapi-boilerplate)
 - [jujumilk3/fastapi-clean-architecture](https://github.com/jujumilk3/fastapi-clean-architecture)
+- [UponTheSky/How to implement a transactional decorator in FastAPI + SQLAlchemy - with reviewing other approaches](https://dev.to/uponthesky/python-post-reviewhow-to-implement-a-transactional-decorator-in-fastapi-sqlalchemy-ein)

app/core/configs.py

@@ -1,16 +1,26 @@
+from enum import Enum
 from typing import Optional
 
 from pydantic import computed_field
 from pydantic_core import MultiHostUrl
 from pydantic_settings import BaseSettings
 
 
+class ENVIRONMENT(Enum):
+    TEST = "TEST"
+    DEV = "DEV"
+    PROD = "PROD"
+
+
 class Configs(BaseSettings):
+    ENV: ENVIRONMENT
+
     # --------- APP SETTINGS --------- #
     PROJECT_NAME: str
     DESCRIPTION: str
     VERSION: str
     PREFIX: str
+    TZ: str = "Asia/Seoul"
 
     # --------- DATABASE SETTINGS --------- #
     DB_TYPE: str

app/core/container.py

@@ -1,16 +1,13 @@
 from dependency_injector.containers import DeclarativeContainer, WiringConfiguration
-from dependency_injector.providers import Factory, Singleton
+from dependency_injector.providers import Factory
 
-from app.core.database import Database
 from app.repositories.users import UserRepository
 from app.services.users import UserService
 
 
 class Container(DeclarativeContainer):
     wiring_config = WiringConfiguration(modules=["app.api.v1.endpoints.users"])
 
-    database = Singleton(Database)
-
-    user_repository = Factory(UserRepository, session=database.provided.session)
+    user_repository = Factory(UserRepository)
 
     user_service = Factory(UserService, user_repository=user_repository)

app/core/database.py

@@ -1,30 +1,87 @@
-from contextlib import asynccontextmanager
-from typing import AsyncIterator
+from contextvars import ContextVar, Token
+from functools import wraps
+from typing import Awaitable, Callable, Optional
 
-from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+from loguru import logger
+from sqlalchemy import NullPool
+from sqlalchemy.ext.asyncio import (
+    AsyncSession,
+    async_scoped_session,
+    async_sessionmaker,
+    create_async_engine,
+)
 
-from app.core.configs import configs
+from app.core.configs import ENVIRONMENT, configs
 from app.models.base import BaseModel
 
 
+class Context:
+    def __init__(self) -> None:
+        self.context: ContextVar[Optional[int]] = ContextVar(
+            "session_context", default=None
+        )
+
+    def get(self) -> int:
+        session_id = self.context.get()
+        if not session_id:
+            raise ValueError("Currently no session is available.")
+        return session_id
+
+    def set(self, session_id: int) -> Token:
+        return self.context.set(session_id)
+
+    def reset(self, context: Token) -> None:
+        self.context.reset(context)
+
+
 class Database:
     def __init__(self) -> None:
-        self.engine = create_async_engine(configs.DATABASE_URI, echo=configs.DB_ECHO)
+        self.context = Context()
+        async_engine_kwargs = {
+            "url": configs.DATABASE_URI,
+            "echo": configs.DB_ECHO,
+        }
+        if configs.ENV == ENVIRONMENT.TEST and configs.DB_DRIVER != "aiosqlite":
+            # NOTE: PyTest 시 event loop 충돌 발생 (related: #19)
+            logger.warning("Using NullPool for async engine")
+            async_engine_kwargs["poolclass"] = NullPool  # type: ignore[assignment]
+        self.engine = create_async_engine(**async_engine_kwargs)  # type: ignore[arg-type]
         self.sessionmaker = async_sessionmaker(
-            bind=self.engine, class_=AsyncSession, expire_on_commit=False
+            bind=self.engine,
+            class_=AsyncSession,
+            autoflush=False,
+            autocommit=False,
+            expire_on_commit=False,
+        )
+        self.scoped_session = async_scoped_session(
+            session_factory=self.sessionmaker,
+            scopefunc=self.context.get,
         )
 
     async def create_all(self) -> None:
+        logger.warning("Create database")
         async with self.engine.begin() as conn:
+            if configs.ENV == ENVIRONMENT.TEST:
+                await conn.run_sync(BaseModel.metadata.drop_all)
             await conn.run_sync(BaseModel.metadata.create_all)
 
-    @asynccontextmanager
-    async def session(self) -> AsyncIterator[AsyncSession]:
-        async with self.sessionmaker() as session:
+    def transactional(self, func: Callable[..., Awaitable]) -> Callable[..., Awaitable]:
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
             try:
-                yield session
-            except Exception:
-                await session.rollback()
-                raise
-            finally:
-                await session.close()
+                session = self.scoped_session()
+                if session.in_transaction():
+                    logger.trace(
+                        f"[Session in transaction]\tID: {database.context.get()}, {self.context=}"
+                    )
+                    return await func(*args, **kwargs)
+                async with session.begin():
+                    response = await func(*args, **kwargs)
+                return response
+            except Exception as error:
+                raise error
+
+        return wrapper
+
+
+database = Database()

app/core/lifespan.py

@@ -6,8 +6,9 @@
 from fastapi import FastAPI
 from loguru import logger
 
-from app.core.configs import configs
+from app.core.configs import ENVIRONMENT, configs
 from app.core.container import Container
+from app.core.database import database
 from app.utils.logging import remove_handler
 
 
@@ -16,9 +17,12 @@ async def lifespan(app: FastAPI):  # pylint: disable=unused-argument
     remove_handler(logging.getLogger("uvicorn.access"))
     remove_handler(logging.getLogger("uvicorn.error"))
     logger.remove()
+    level = 0
+    if configs.ENV == ENVIRONMENT.PROD:
+        level = 20
     logger.add(
         sys.stderr,
-        level=0,
+        level=level,
         format="<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> <bg #800a0a>"
         + time.tzname[0]
         + "</bg #800a0a> | <level>{level: <8}</level> | <fg #800a0a>{name}</fg #800a0a>:<fg #800a0a>{function}</fg #800a0a>:<fg #800a0a>{line}</fg #800a0a> - <level>{message}</level>",
@@ -27,10 +31,9 @@ async def lifespan(app: FastAPI):  # pylint: disable=unused-argument
     # logging.getLogger("uvicorn.access").addHandler(LoguruHandler())
     # logging.getLogger("uvicorn.error").addHandler(LoguruHandler())
 
-    container = Container()
+    logger.info(f"{configs.ENV=}")
     if configs.DB_TABLE_CREATE:
-        logger.warning("Create database")
-        database = container.database()
         await database.create_all()
+    app.container = Container()  # type: ignore[attr-defined]
 
     yield

app/core/middlewares.py

@@ -4,6 +4,7 @@
 from loguru import logger
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
 
+from app.core.database import database
 from app.utils.logging import ANSI_BG_COLOR, ANSI_STYLE, ansi_format
 
 
@@ -57,3 +58,18 @@ async def dispatch(
             f"[IP: {ip}] [URL: {url}] [Method: {method}] [Status: {status} (Elapsed Time: {elapsed_time})]"
         )
         return response
+
+
+class SessionMiddleware(BaseHTTPMiddleware):
+    async def dispatch(
+        self, request: Request, call_next: RequestResponseEndpoint
+    ) -> Response:
+        try:
+            context = database.context.set(session_id=hash(request))
+            logger.trace(f"[Session Start]\tID: {database.context.get()}, {context=}")
+            response = await call_next(request)
+        finally:
+            await database.scoped_session.remove()
+            logger.trace(f"[Session End]\tID: {database.context.get()}, {context=}")
+            database.context.reset(context=context)
+        return response

app/exceptions/base.py

@@ -4,3 +4,11 @@
 class CoreException(abc.ABC, Exception):
     status: int
     message: str
+
+    def __str__(self) -> str:
+        return (
+            f"[{self.__class__.__name__}] status={self.status}, message={self.message}"
+        )
+
+    def __repr__(self) -> str:
+        return f"[{self.__class__.__name__}] {self.message}"

app/exceptions/database.py

@@ -3,6 +3,16 @@
 from app.exceptions.base import CoreException
 
 
-class EntityNotFound(CoreException):
+class DatabaseException(CoreException):
+    status: int
+    message: str
+
+
+class EntityAlreadyExists(DatabaseException):
+    status: int = status.HTTP_409_CONFLICT
+    message: str = "Entity already exists in the database."
+
+
+class EntityNotFound(DatabaseException):
     status: int = status.HTTP_404_NOT_FOUND
     message: str = "Entity not found in the database."

app/exceptions/handlers.py

@@ -19,14 +19,13 @@ async def global_exception_handler(request: Request, exc: Exception) -> JSONResp
     )
 
 
-async def business_exception_handler(
-    request: Request, exc: CoreException
+async def core_exception_handler(
+    request: Request, exc: CoreException  # pylint: disable=unused-argument
 ) -> JSONResponse:
-    logger.error(f"{request=}, {exc=}")
-    name = exc.__class__.__name__
+    logger.error(exc)
     return JSONResponse(
-        content=APIResponse.error(
-            status=exc.status, message=f"[{name}] {exc.message}"
-        ).model_dump(mode="json"),
+        content=APIResponse.error(status=exc.status, message=repr(exc)).model_dump(
+            mode="json"
+        ),
         status_code=exc.status,
     )