FastAPI Patterns

Best practices and patterns for FastAPI projects with SQLAlchemy 2.0 and dependency-injector.

Triggers

Use this skill when the user:

• Creates a new FastAPI project or module
• Asks about FastAPI application structure
• Works with SQLAlchemy models and repositories
• Configures Dependency Injection
• Creates DTOs (Pydantic models)

Layer Architecture

Router → Service → Repository → Database
   ↓        ↓          ↓
  DTO     DTO     Model/DTO

• Router — HTTP endpoints, request validation
• Service — business logic, orchestration
• Repository — database operations, CRUD operations

Modular Structure

src/
├── core/
│   ├── config.py         # Settings (pydantic-settings)
│   ├── database.py       # Engine, Base, session
│   ├── container.py      # DI Container
│   ├── dependencies.py   # FastAPI dependencies
│   ├── exceptions.py     # Custom exceptions
│   └── repositories.py   # BaseRepository
├── modules/
│   ├── users/
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── dto.py
│   │   ├── repositories.py
│   │   ├── services.py
│   │   └── routers.py
│   └── orders/
│       └── ...
└── main.py

BaseRepository with Generics

More details: ${CLAUDE_PLUGIN_ROOT}/skills/fastapi-patterns/references/repository.md

from typing import Generic, TypeVar
from collections.abc import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

ModelT = TypeVar("ModelT")
CreateDTOT = TypeVar("CreateDTOT")
ReadDTOT = TypeVar("ReadDTOT")


class BaseRepository(Generic[ModelT, CreateDTOT, ReadDTOT]):
    """Base repository with generic CRUD operations."""

    def __init__(self, session: AsyncSession, model: type[ModelT]) -> None:
        self._session = session
        self._model = model

    async def get_all(self) -> Sequence[ReadDTOT]:
        result = await self._session.execute(select(self._model))
        return result.scalars().all()

    async def save(self, data: ModelT | CreateDTOT) -> ReadDTOT:
        if isinstance(data, self._model):
            entity = data
        else:
            entity = self._model(**data.model_dump())
        self._session.add(entity)
        await self._session.flush()
        return entity

DTOs with Pydantic v2

More details: ${CLAUDE_PLUGIN_ROOT}/skills/fastapi-patterns/references/dto.md

from pydantic import BaseModel, ConfigDict


class BaseDTO(BaseModel):
    """Base DTO with ORM mode."""

    model_config = ConfigDict(
        from_attributes=True,
        populate_by_name=True,
        str_strip_whitespace=True,
    )


class UserCreateDTO(BaseDTO):
    email: str
    name: str


class UserReadDTO(BaseDTO):
    id: int
    email: str
    name: str

Dependency Injection

More details: ${CLAUDE_PLUGIN_ROOT}/skills/fastapi-patterns/references/di.md

# container.py
from dependency_injector import containers, providers


class Container(containers.DeclarativeContainer):
    config = providers.Configuration()

    session_maker = providers.Singleton(
        async_sessionmaker,
        bind=engine,
    )

    user_repository = providers.Factory(
        UserRepository,
        session=session_maker,
    )

    user_service = providers.Factory(
        UserService,
        repository=user_repository,
    )


# routers.py
from dependency_injector.wiring import inject, Provide


@router.get("/users/{id}")
@inject
async def get_user(
    id: int,
    service: Annotated[UserService, Depends(Provide[Container.user_service])],
) -> UserReadDTO:
    return await service.get_by_id(id)

Exception Handling

More details: ${CLAUDE_PLUGIN_ROOT}/skills/fastapi-patterns/references/exceptions.md

# exceptions.py
class AppError(Exception):
    """Base application error."""

    status_code: int = 500
    detail: str = "Internal server error"

    def __init__(self, detail: str | None = None) -> None:
        self.detail = detail or self.detail


class NotFoundError(AppError):
    status_code = 404
    detail = "Resource not found"


class ConflictError(AppError):
    status_code = 409
    detail = "Resource already exists"


# handlers.py
async def app_exception_handler(request: Request, exc: AppError) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
    )


# main.py
app.add_exception_handler(AppError, app_exception_handler)

Transaction Management

# Via session_maker context manager
async def create_order(self, data: OrderCreateDTO) -> OrderReadDTO:
    async with self._session_maker() as session:
        order = Order(**data.model_dump())
        session.add(order)
        await session.commit()
        return OrderReadDTO.model_validate(order)

# Or via repository
async def create_with_items(self, data: OrderCreateDTO) -> OrderReadDTO:
    order = await self._order_repo.save(data)
    for item in data.items:
        await self._item_repo.save(OrderItem(order_id=order.id, **item))
    await self._session.commit()
    return order

Plugin Commands

• /fastapi:module <name> — create a complete module
• /fastapi:endpoint <method> <path> <module> — create an endpoint
• /fastapi:dto <model> — create DTOs from model