Harleypig Dotfiles Fastapi Patterns

FastAPI Patterns

Version: v1.0.0

Concrete, copy-adaptable recipes for FastAPI services with Pydantic v2 and SQLAlchemy 2.0. rules/fastapi.md is the source of truth for the conventions (one response_model per endpoint, lifespan for startup, HTTPException at the edge, pydantic-settings for config) — read it first; this skill only adds the deeper how. For the persistence layer see the sqlalchemy-patterns skill; this skill stops at the repository boundary.

This skill is idiom-specific: it uses FastAPI-native Depends(...) with a session-per-request dependency. It deliberately does not use the dependency-injector library that some upstream sources lean on (see SOURCE.md) — match the surrounding code.

When to reach for this

Writing endpoints, request/response schemas, a service or repository layer, or API-level error handling — and the one-liners in rules/fastapi.md aren't enough detail.

DTOs (Pydantic v2)

Base config once; every DTO inherits ORM-mode and whitespace stripping:

from pydantic import BaseModel, ConfigDict

class BaseDTO(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,      # build DTOs straight from ORM objects
        populate_by_name=True,     # accept field name or alias
        str_strip_whitespace=True,
    )

Create / Read / Update split. Create omits server-set fields (id, timestamps); Read includes them; Update makes every field optional and is applied with exclude_unset so absent keys are left untouched:

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

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

class UserUpdate(BaseDTO):
    email: str | None = None
    name: str | None = None

# in the service: patch only the fields actually sent
data = payload.model_dump(exclude_unset=True)
for field, value in data.items():
    setattr(user, field, value)

Field vs model validators. @field_validator for one field; @model_validator(mode="after") for cross-field rules:

from pydantic import field_validator, model_validator

class SignUp(BaseDTO):
    email: str
    password: str
    password_confirm: str

    @field_validator("email")
    @classmethod
    def normalize_email(cls, v: str) -> str:
        return v.lower()

    @model_validator(mode="after")
    def passwords_match(self) -> "SignUp":
        if self.password != self.password_confirm:
            raise ValueError("passwords do not match")
        return self

Computed and nested. @computed_field derives values at serialization; nest Read DTOs directly:

from pydantic import computed_field

class OrderRead(BaseDTO):
    id: int
    items: list[OrderItemRead]

    @computed_field
    @property
    def total(self) -> float:
        return sum(i.price * i.qty for i in self.items)

Enums serialize for free when the member type is str:

from enum import Enum

class OrderStatus(str, Enum):
    pending = "pending"
    shipped = "shipped"

class OrderRead(BaseDTO):
    status: OrderStatus

camelCase at the boundary, snake_case inside — alias the field and dump by_alias only at the JSON edge:

from pydantic import Field

class Profile(BaseDTO):
    full_name: str = Field(alias="fullName")

Dependencies and serialization (native FastAPI)

Annotated dependency aliases — declare once, reuse everywhere; the type survives outside FastAPI too:

from typing import Annotated
from fastapi import Depends

SessionDep = Annotated[Session, Depends(get_session)]
CurrentUserDep = Annotated[User, Depends(get_current_user)]

@router.get("/me")
async def me(user: CurrentUserDep) -> UserRead:
    return user

Return type drives serialization. Annotate the return; Pydantic v2 serializes in Rust and documents the schema automatically. Reach for response_model only when the wire shape must differ from the returned object (e.g. filtering secret fields).

async def only when you await async I/O. A plain def path operation runs in the threadpool, so blocking work there won't stall the event loop — but blocking calls inside an async def will. Pick the keyword by what the body actually does.

Router-level metadata — set prefix/tags/dependencies on the APIRouter, not on every include_router(...) call, so they live next to the endpoints they describe.

Repository pattern

A small generic base carries the typed CRUD; concrete repos add domain queries. The session arrives by constructor injection (from a Depends session dependency). The repository flushes; it does not commit — the transaction boundary belongs to the service or the request dependency, so a unit of work can span several repositories:

from collections.abc import Sequence
from typing import Generic, TypeVar
from sqlalchemy import select
from sqlalchemy.orm import Session

ModelT = TypeVar("ModelT")

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

    def get(self, id_: int) -> ModelT | None:
        return self._session.get(self._model, id_)

    def list(self) -> Sequence[ModelT]:
        return self._session.execute(select(self._model)).scalars().all()

    def add(self, entity: ModelT) -> ModelT:
        self._session.add(entity)
        self._session.flush()   # assign PKs/defaults; no commit here
        return entity

Concrete repos add the domain-specific reads and prevent N+1 with eager loading:

from sqlalchemy.orm import selectinload

class UserRepository(BaseRepository[User]):
    def __init__(self, session: SessionDep) -> None:
        super().__init__(session, User)

    def by_email(self, email: str) -> User | None:
        stmt = select(User).where(User.email == email)
        return self._session.execute(stmt).scalar_one_or_none()

    def with_orders(self, id_: int) -> User | None:
        stmt = (
            select(User)
            .where(User.id == id_)
            .options(selectinload(User.orders))
        )
        return self._session.execute(stmt).scalar_one_or_none()

(SQLAlchemy 2.0 model/session specifics — Mapped, the session dependency, the SQLite FK pragma — live in the sqlalchemy-patterns skill and rules/sqlalchemy.md.)

Exception → HTTP mapping

Raise meaning-bearing errors in services; translate to HTTP once, at the edge, with a registered handler — so business code stays free of HTTPException:

# errors.py
class AppError(Exception):
    status_code = 500
    detail = "Internal server error"

    def __init__(self, detail: str | None = None) -> None:
        if detail is not None:
            self.detail = detail

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

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

# wire once, in the app factory / lifespan setup
from fastapi import Request
from fastapi.responses import JSONResponse

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

app.add_exception_handler(AppError, handle_app_error)

Services then express intent without knowing about HTTP:

def get_user(self, id_: int) -> User:
    user = self._repo.get(id_)
    if user is None:
        raise NotFoundError(f"user {id_} not found")
    return user

This is the depth behind the rule's "translate to HTTP only at the endpoint/dependency layer" — a handler is the clean alternative to a try/except HTTPException in every endpoint.

Troubleshooting

When an endpoint misbehaves in a way that doesn't reduce to obvious logic — a schema that won't validate, a response serializing wrong, a type that looks right but isn't — run the qa-check skill before deep-diving. Its format (ruff format), lint (ruff check), and type-check pass surfaces the usual culprits (a mis-typed Mapped, a DTO field that doesn't match the model, an un-awaited coroutine) faster than reading the traceback. Fix what it finds, then resume.

Provenance

Adapted from upstream sources (ideas drawn, re-rendered in this repo's native idiom), not vendored. See SOURCE.md for repos, SHAs, and the upstream-check procedure.

Source: harleypig/dotfiles — distributed by TomeVault.