fastapi

# FastAPI & modern Python services The single authoritative skill for building, reviewing, testing, securing and shipping an async FastAPI service on Python 3.12+. The mental model: **the app is a thin async HTTP layer over typed dependencies, a service/repository core, and explicit DB sessions. Routes validate and delegate; they never own business logic, raw SQL, or secrets.** Pinned stack: Python 3.12+, FastAPI 0.136+, Starlette 1.0+ (FastAPI 0.136 requires it; avoid <1.0.1, GHSA-86qp-5c8j-p5mr), Pydantic v2 (2.7+) + pydantic-settings 2.x, SQLAlchemy 2.0 async, Alembic 1.13+, asyncpg 0.30 / psycopg 3, httpx 0.28+, pytest 8 + pytest-asyncio 1.0+ (`asyncio_mode=auto`), ruff 0.7+, mypy 1.13+ strict, uv 0.5+, uvicorn 0.32+ / gunicorn 23+ + uvicorn-worker 0.3+, PyJWT 2.10+, argon2-cffi 23+, pip-audit 2.7+, PostgreSQL 16. (All lower bounds; install the latest in each line.) > **⚠️ SDD new-feature gate — read this first.** If this skill fired on a **new, non-trivial feature or behaviour change** and there is **no approved spec + plan** under `02-DOCS/wiki/sdd/`, STOP — do **not** write feature code yet. Hand off to `../specify/SKILL.md` first: it runs brainstorm → spec → plan → tasks before any code, then routes back here once the plan is approved. Build here directly only for a genuinely one-line / low-risk change. Method: `../sdd/SKILL.md`. Out of scope, and where it goes instead: [`django`](../django/SKILL.md) for Django; Flask / sync WSGI, notebooks and CLI-only scripts (no skill); language-level Python, typing and packaging → [`python`](../python/SKILL.md); framework-agnostic REST contracts — status codes, URL naming, versioning, cursor vs offset → [`api-design`](../api-design/SKILL.md) (this skill covers their FastAPI *implementation*); engine-level schema, indexing, EXPLAIN, zero-downtime migrations, PgBouncer → [`postgresdb`](../postgresdb/SKILL.md); language-agnostic injection / secret / authz theory → [`secure-coding`](../secure-coding/SKILL.md); Dockerfile, Compose and CI/CD mechanics → [`deployment`](../deployment/SKILL.md) (this skill keeps only a Docker *note*). ## Decision rules 1. `async def` for any I/O route + async drivers (asyncpg, httpx); never `requests`/`psycopg2`/blocking calls on the loop (offload via `await anyio.to_thread.run_sync`). 2. Three Pydantic models per resource — `XCreate`/`XUpdate`/`XResponse` (`from_attributes=True`); responses never leak hashes/tokens/internal flags. 3. Request-scoped resources via `Annotated[T, Depends(...)]`, never built inline — so tests can override them. 4. One DB session per request via `get_db` (commit-on-success / rollback-on-exception); handlers never commit. 5. One error envelope `{"error":{"code","message","details?"}}` via centralized handlers; never leak stack traces / SQL. 6. Settings from `pydantic-settings` (`BaseSettings`), never scattered `os.getenv`. 7. Validate JWT `exp`/`iss`/`aud` and pin `algorithms=["RS256"|"HS256"]` explicitly. 8. Tests: `ASGITransport` + `dependency_overrides` on a transactional DB; CI gates on `ruff`, `mypy --strict`, `pytest --cov`, `pip-audit`. ## Project layout ```text app/ ├── main.py # create_app() factory + lifespan; app = create_app() ├── core/ │ ├── config.py # Settings(BaseSettings) + get_settings() │ ├── security.py # hashing, JWT encode/decode │ └── logging.py # structlog / JSON logging setup ├── api/ │ ├── deps.py # get_db, get_current_user, Pagination, require_roles │ └── routers/ │ ├── users.py │ └── health.py ├── schemas/ # Pydantic v2 models (Create/Update/Response) │ └── user.py ├── models/ # SQLAlchemy 2.0 DeclarativeBase models │ └── user.py ├── db/ │ ├── base.py # engine, async_sessionmaker, Base │ └── repository.py # generic async Repository[ModelT] ├── services/ # business logic (no FastAPI imports) │ └── user_service.py ├── exceptions.py # AppError hierarchy + register_exception_handlers tests/ # pytest-asyncio + ASGITransport alembic/ # async env.py + versions/ pyproject.toml # ruff + mypy strict + pytest config ``` Routers stay thin, services hold the logic, the repository/CRUD layer owns persistence. ## Application factory + lifespan ```python from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from app.api.routers import health, users from app.core.config import Settings, get_settings from app.db.base import engine from app.exceptions import register_exception_handlers # Routers paired with their mount prefix + OpenAPI tag, declared once so the factory # stays a flat loop instead of a wall of include_router() calls. ROUTERS = ( (health.router, "/health", "health"), (users.router, "/api/v1/users", "users"), ) @asynccontextmanager async def lifespan(_app: FastAPI): # Open pools/caches on startup (here), never at import time, so importing the module has # no side effects (tests and Alembic import it freely). yield await engine.dispose() # release pooled DB connections so workers exit cleanly def _install_cors(app: FastAPI, settings: Settings) -> None: if not settings.cors_origins: return # no browser clients configured -> skip the middleware entirely app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, # explicit per-env list, never ["*"] with creds allow_credentials=True, allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"], allow_headers=["Authorization", "Content-Type"], ) def create_app(settings: Settings | None = None) -> FastAPI: settings = settings or get_settings() app = FastAPI(title=settings.api_title, version=settings.api_version, lifespan=lifespan) register_exception_handlers(app) _install_cors(app, settings) for router, prefix, tag in ROUTERS: app.include_router(router, prefix=prefix, tags=[tag]) return app app = create_app() ``` Accepting an optional `settings` argument lets tests build the app with overridden config without touching the `get_settings` cache. **Bad** = `allow_origins=["*"]` with `allow_credentials=True` — browsers reject it and Starlette refuses to echo `*` for credentialed requests. → [`references/production.md`](references/production.md) for proxy headers / logging wiring at startup. To inject servers / security schemes / a logo into the generated OpenAPI doc, assign a custom builder to `app.openapi` inside `create_app()`. → [`references/production.md`](references/production.md) (Customizing the OpenAPI schema). ## Configuration (pydantic-settings) ```python from functools import lru_cache from pydantic import PostgresDsn, SecretStr from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_", extra="ignore") api_title: str = "Service API" api_version: str = "1.0.0" environment: str = "development" database_url: PostgresDsn jwt_secret: SecretStr jwt_algorithm: str = "HS256" jwt_issuer: str = "service-api" jwt_audience: str = "service-clients" access_token_ttl_seconds: int = 900 cors_origins: list[str] = [] @lru_cache def get_settings() -> Settings: return Settings() # type: ignore[call-arg] # values come from env/.env ``` **Bad** = `DB_URL = os.environ["DB_URL"]` at import time (crashes on import, untyped, unmockable). **Good** = inject `get_settings` as a dependency so tests override it. ## Pydantic v2 models (Create/Update/Response split) ```python from datetime import datetime from typing import Annotated from uuid import UUID from pydantic import BaseModel, ConfigDict, EmailStr, Field, computed_field # Reusable constrained types keep the same rule in one place across the three models. FullName = Annotated[str, Field(min_length=1, max_length=100)] RawPassword = Annotated[str, Field(min_length=12, max_length=128)] class UserInput(BaseModel): """Fields a client may send. Create/Update narrow this; Response never inherits it.""" email: EmailStr full_name: FullName class UserCreate(UserInput): password: RawPassword class UserUpdate(BaseModel): # Every field optional: a PATCH sends only what changes. email: EmailStr | None = None full_name: FullName | None = None class UserResponse(BaseModel): model_config = ConfigDict(from_attributes=True) # populate straight off ORM attributes id: UUID email: EmailStr full_name: str created_at: datetime @computed_field # type: ignore[prop-decorator] @property def label(self) -> str: return f"{self.full_name} <{self.email}>" ``` v2 migration cheats: use `.model_dump()` not `.dict()`; `.model_validate(obj)` not `.from_orm()`; `model_config = ConfigDict(...)` not class `Config`; `field_validator`/`model_validator` not `@validator`/`@root_validator`. **Bad** = a response model with `hashed_password: str` (leaks the hash). **Good** = the `UserResponse` above (no secret fields). → [`references/security.md`](references/security.md). ## Dependency injection ```python from collections.abc import AsyncIterator from dataclasses import dataclass from typing import Annotated from fastapi import Depends, Query from sqlalchemy.ext.asyncio import AsyncSession from app.db.base import async_session_factory async def get_db() -> AsyncIterator[AsyncSession]: session = async_session_factory() try: yield session await session.commit() # commit only if the handler returned without raising except Exception: await session.rollback() # any error (incl. HTTP exceptions) unwinds the txn raise finally: await session.close() # always release the connection back to the pool DbSession = Annotated[AsyncSession, Depends(get_db)] @dataclass(frozen=True) class Pagination: limit: int offset: int def get_pagination( limit: Annotated[int, Query(ge=1, le=100)] = 50, offset: Annotated[int, Query(ge=0)] = 0, ) -> Pagination: return Pagination(limit=limit, offset=offset) PageParams = Annotated[Pagination, Depends(get_pagination)] ``` → [`references/database.md`](references/database.md) for `async_session_factory` wiring; → [`references/security.md`](references/security.md) for `get_current_user` and `require_roles`. ## Routers & endpoints ```python from fastapi import APIRouter, Response, status from app.api.deps import CurrentUser, DbSession, PageParams from app.schemas.user import UserCreate, UserResponse from app.services import user_service router = APIRouter() @router.get("", response_model=list[UserResponse]) async def list_users(db: DbSession, page: PageParams) -> list[UserResponse]: return await user_service.list_users(db, limit=page.limit, offset=page.offset) @router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED) async def create_user(payload: UserCreate, db: DbSession, response: Response) -> UserResponse: user = await user_service.create_user(db, payload) response.headers["Location"] = f"/api/v1/users/{user.id}" return user ``` **Bad** = hashing the password + building `select()` + business rules inline in the route. **Good** = `await user_service.create_user(db, payload)` (route stays thin). `CurrentUser` is defined in `references/security.md`. ## Error handling & envelope ```python from fastapi import FastAPI, Request, status from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse from app.core.logging import logger class AppError(Exception): def __init__(self, message: str, code: str, status_code: int = 500, details: list[dict] | None = None) -> None: super().__init__(message) self.message = message self.code = code self.status_code = status_code self.details = details or [] class NotFoundError(AppError): def __init__(self, resource: str, ident: str) -> None: super().__init__(f"{resource} not found: {ident}", "not_found", 404) def register_exception_handlers(app: FastAPI) -> None: @app.exception_handler(AppError) async def _app_error(request: Request, exc: AppError) -> JSONResponse: return JSONResponse( status_code=exc.status_code, content={"error": {"code": exc.code, "message": exc.message, "details": exc.details}}, ) @app.exception_handler(RequestValidationError) async def _validation(request: Request, exc: RequestValidationError) -> JSONResponse: details = [{"field": ".".join(map(str, e["loc"][1:])), "message": e["msg"], "code": e["type"]} for e in exc.errors()] return JSONResponse( status_code=422, content={"error": {"code": "validation_error", "message": "Request validation failed", "details": details}}, ) @app.exception_handler(Exception) async def _unhandled(request: Request, exc: Exception) -> JSONResponse: logger.exception("unhandled_error", path=request.url.path) return JSONResponse( status_code=500,