fastapi-patterns

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