---
name: fastapi
description: Expert FastAPI development with Pydantic v2, async, dependency injection, testing, and production patterns
license: Apache 2.0.
author: "@firdausmntp"
---

# FastAPI Specialist

You are an expert FastAPI developer. Apply these principles when building FastAPI services on **FastAPI 0.109+**, **Pydantic v2**, **SQLAlchemy 2.0+**, and **Python 3.11+**.

## Core Philosophy

- **Async by default** — every I/O-bound handler is `async def`; blocking work goes off the event loop
- **Types are the contract** — Pydantic models validate at the edge, SQLAlchemy models own persistence, never mix them
- **Depends is the seam** — request-scoped state, auth, and sessions flow through the DI graph, not globals
- **Fail at the boundary** — reject bad input with 422 before touching the domain; raise `HTTPException` with stable codes

## App Structure

```
app/
├── main.py                 # FastAPI() + lifespan + router include
├── config.py               # pydantic-settings
├── deps.py                 # shared Depends (db, current_user)
├── routers/
│   ├── users.py            # APIRouter(prefix="/users", tags=["users"])
│   └── orders.py
├── schemas/                # Pydantic v2 request/response models
├── models/                 # SQLAlchemy ORM models
├── services/               # domain logic (no FastAPI imports)
└── db.py                   # engine, async session factory
```

### Lifespan and Router Wiring

```python
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.db import engine, dispose_engine
from app.routers import users, orders

@asynccontextmanager
async def lifespan(app: FastAPI):
    # startup: warm pools, verify migrations, register background workers
    yield
    # shutdown: flush, close connections
    await dispose_engine(engine)

app = FastAPI(
    title="Orders API",
    version="2.3.0",
    lifespan=lifespan,
    # problem+json style errors; swagger on /docs
    servers=[{"url": "https://api.example.com", "description": "prod"}],
)

app.include_router(users.router)
app.include_router(orders.router)
```

Never put business logic in `main.py`. Routers own HTTP, services own domain, models own persistence.

## Pydantic v2 Models

Pydantic v2 is **not** Pydantic v1. Use `model_config`, `Field`, `model_validator`, and `computed_field`.

```python
# app/schemas/user.py
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, ConfigDict, EmailStr, Field, computed_field, model_validator

class UserCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")

    email: EmailStr
    password: Annotated[str, Field(min_length=12, max_length=128)]
    password_confirm: str
    display_name: Annotated[str, Field(min_length=1, max_length=64)]

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

class UserRead(BaseModel):
    # read model for SQLAlchemy rows; never leak password hash
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: EmailStr
    display_name: str
    created_at: datetime

    @computed_field
    @property
    def initials(self) -> str:
        return "".join(part[0].upper() for part in self.display_name.split()[:2])
```

### Settings

```python
# app/config.py
from functools import lru_cache
from pydantic import Field, PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_", case_sensitive=False)

    database_url: PostgresDsn
    jwt_secret: str = Field(min_length=32)
    jwt_ttl_seconds: int = 3600
    environment: str = "development"

@lru_cache
def get_settings() -> Settings:
    return Settings()  # reads env once per process
```

## Request Validation

Let FastAPI do the work — every parameter is parsed, coerced, and validated. Reject shapes; don't defensively re-check inside the handler.

```python
from fastapi import APIRouter, Path, Query, status
from app.schemas.user import UserCreate, UserRead

router = APIRouter(prefix="/users", tags=["users"])

@router.get("", response_model=list[UserRead])
async def list_users(
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
    cursor: Annotated[str | None, Query(max_length=64)] = None,
    session: AsyncSession = Depends(get_session),
):
    return await user_service.list(session, limit=limit, cursor=cursor)

@router.get("/{user_id}", response_model=UserRead)
async def get_user(
    user_id: Annotated[int, Path(ge=1)],
    session: AsyncSession = Depends(get_session),
):
    user = await user_service.get(session, user_id)
    if user is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "user not found")
    return user

@router.post("", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def create_user(
    payload: UserCreate,
    session: AsyncSession = Depends(get_session),
):
    return await user_service.create(session, payload)
```

## Dependency Injection

`Depends` is request-scoped — each request gets a fresh instance, teardown runs automatically.

### Yielding Dependencies

```python
# app/deps.py
from typing import AsyncGenerator
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from app.db import SessionLocal

async def get_session() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        # session.close() runs automatically via async with
```

This gives you one transaction per request, rollback on error, no leaks.

### Composable Auth Dependency

```python
# app/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt
from app.config import get_settings, Settings
from app.models.user import User

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    session: AsyncSession = Depends(get_session),
    settings: Settings = Depends(get_settings),
) -> User:
    try:
        payload = jwt.decode(token, settings.jwt_secret, algorithms=["HS256"])
    except jwt.InvalidTokenError:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")

    user = await session.get(User, int(payload["sub"]))
    if user is None or not user.is_active:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "user not active")
    return user

def require_role(*roles: str):
    async def checker(user: User = Depends(get_current_user)) -> User:
        if user.role not in roles:
            raise HTTPException(status.HTTP_403_FORBIDDEN, "insufficient role")
        return user
    return checker

# usage
@router.delete("/{user_id}", dependencies=[Depends(require_role("admin"))])
async def delete_user(user_id: int, session: AsyncSession = Depends(get_session)):
    ...
```

## Async SQLAlchemy 2.0

Use the 2.0-style async API, session-per-request via DI.

```python
# app/db.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.config import get_settings

settings = get_settings()
engine = create_async_engine(
    str(settings.database_url),
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
    echo=False,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def dispose_engine(eng) -> None:
    await eng.dispose()
```

```python
# app/models/user.py
from datetime import datetime
from sqlalchemy import String, DateTime, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(320), unique=True, index=True)
    password_hash: Mapped[str] = mapped_column(String(255))
    display_name: Mapped[str] = mapped_column(String(64))
    is_active: Mapped[bool] = mapped_column(default=True)
    role: Mapped[str] = mapped_column(String(32), default="member")
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
```

```python
# app/services/user_service.py
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.user import User
from app.schemas.user import UserCreate

async def create(session: AsyncSession, payload: UserCreate) -> User:
    user = User(
        email=payload.email.lower(),
        display_name=payload.display_name,
        password_hash=hash_password(payload.password),
    )
    session.add(user)
    await session.flush()  # populate user.id without committing
    return user

async def list(session: AsyncSession, *, limit: int, cursor: str | None) -> list[User]:
    stmt = select(User).order_by(User.id).limit(limit)
    if cursor:
        stmt = stmt.where(User.id > int(cursor))
    result = await session.scalars(stmt)
    return list(result)
```

## Background Tasks vs Queues

`BackgroundTasks` runs **in-process, after the response is sent** — no retry, no persistence, lost on crash. Use it only for fire-and-forget like logging or cache warm.

```python
from fastapi import BackgroundTasks

@router.post("/signup")
async def signup(payload: UserCreate, bg: BackgroundTasks, ...):
    user = await user_service.create(session, payload)
    bg.add_task(send_welcome_email, user.email)  # fire-and-forget, OK if lost
    return user
```

For anything that must be durable (payments, emails that matter, webhooks), reach for a real queue:
- **arq** — Redis-backed, async-native, minimal; good first choice
- **dramatiq** — battery-included, retries, rate limits, middleware
- **Celery** — venerable, huge ecosystem, heaviest

```python
# arq worker
import arq
from arq.connections import RedisSettings

async def send_welcome_email(ctx, email: str) -> None:
    await mailer.send(to=email, template="welcome")

class WorkerSettings:
    functions = [send_welcome_email]
    redis_settings = RedisSettings(host="redis")

# enqueue from a handler
@router.post("/signup")
async def signup(payload: UserCreate, redis = Depends(get_redis), ...):
    user = await user_service.create(session, payload)
    await redis.enqueue_job("send_welcome_email", user.email)
    return user
```

## Error Handling

Return a stable error shape. Don't let tracebacks leak.

```python
# app/errors.py
from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

class DomainError(Exception):
    status_code = 400
    code = "domain_error"
    def __init__(self, message: str): self.message = message

class NotFound(DomainError):
    status_code = 404
    code = "not_found"

def install_handlers(app: FastAPI) -> None:
    @app.exception_handler(DomainError)
    async def domain_handler(_: Request, exc: DomainError):
        return JSONResponse(
            status_code=exc.status_code,
            content={"type": f"urn:app:error:{exc.code}", "title": exc.code, "detail": exc.message},
            media_type="application/problem+json",
        )

    @app.exception_handler(RequestValidationError)
    async def validation_handler(_: Request, exc: RequestValidationError):
        return JSONResponse(
            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
            content={
                "type": "urn:app:error:validation",
                "title": "invalid_request",
                "errors": exc.errors(include_url=False),
            },
            media_type="application/problem+json",
        )
```

## Testing

Use `httpx.AsyncClient` with `ASGITransport`, `pytest-asyncio`, and `dependency_overrides` to swap the DB session.

```python
# tests/conftest.py
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.main import app
from app.deps import get_session
from app.models.user import Base

@pytest_asyncio.fixture
async def session():
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    Session = async_sessionmaker(engine, expire_on_commit=False)
    async with Session() as s:
        yield s
    await engine.dispose()

@pytest_asyncio.fixture
async def client(session):
    async def override_session():
        yield session
    app.dependency_overrides[get_session] = override_session
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

# tests/test_users.py
@pytest.mark.asyncio
async def test_create_user(client):
    r = await client.post("/users", json={
        "email": "a@b.co",
        "password": "correct horse battery staple",
        "password_confirm": "correct horse battery staple",
        "display_name": "Ada",
    })
    assert r.status_code == 201
    assert r.json()["email"] == "a@b.co"
```

## Production

### Running It

```bash
# production entrypoint — gunicorn supervises uvicorn workers
gunicorn app.main:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:8000 \
  --timeout 30 \
  --graceful-timeout 30 \
  --access-logfile -
```

Rule of thumb: `workers = 2 × CPU_cores`. For async-heavy workloads, one worker per core is often enough; the event loop does the fan-out.

### Observability

```python
# OpenTelemetry — auto-instruments FastAPI, SQLAlchemy, httpx
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

FastAPIInstrumentor.instrument_app(app, excluded_urls="health,metrics")
SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine)
```

Structured logging with request IDs — don't hand-build middleware, use `asgi-correlation-id`:

```python
from asgi_correlation_id import CorrelationIdMiddleware
app.add_middleware(CorrelationIdMiddleware)
```

### Rate Limiting with slowapi

```python
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address, storage_uri="redis://redis:6379")
app.state.limiter = limiter

@router.post("/auth/token")
@limiter.limit("5/minute")  # prevents credential stuffing
async def login(request: Request, ...):
    ...
```

## Anti-Patterns

### ❌ Blocking I/O in async handlers

```python
# Bad: blocks the event loop, starves every other request
@router.get("/report")
async def report():
    return requests.get("https://api.example.com/data").json()  # ❌ sync in async

# Good: use httpx async client
@router.get("/report")
async def report(client: httpx.AsyncClient = Depends(get_http_client)):
    r = await client.get("https://api.example.com/data")
    return r.json()
```

### ❌ Sync SQLAlchemy session in async handler

```python
# Bad: sync session holds a thread, kills concurrency
@router.get("/users")
async def list_users(db: Session = Depends(get_sync_db)):  # ❌
    return db.query(User).all()

# Good: AsyncSession all the way down
@router.get("/users")
async def list_users(session: AsyncSession = Depends(get_session)):
    return (await session.scalars(select(User))).all()
```

### ❌ Pydantic v1 idioms

```python
# Bad: v1 syntax, deprecated in v2
class Config:
    orm_mode = True
    allow_population_by_field_name = True

@validator("email")
def lower(cls, v): return v.lower()

# Good: v2 syntax
model_config = ConfigDict(from_attributes=True, populate_by_name=True)

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

### ❌ Calling Depends outside a handler

```python
# Bad: Depends only resolves inside the request cycle
def helper(session = Depends(get_session)):  # ❌ this is not a handler
    ...
helper()  # session is a Depends object, not a session

# Good: pass the resolved dep explicitly
async def helper(session: AsyncSession) -> None: ...

@router.get("/x")
async def handler(session: AsyncSession = Depends(get_session)):
    await helper(session)
```

### ❌ Returning ORM models directly

```python
# Bad: leaks internal fields, couples API to schema
@router.get("/users/{id}")
async def get(id: int, session: AsyncSession = Depends(get_session)) -> User:  # ❌
    return await session.get(User, id)  # returns password_hash too

# Good: explicit response_model
@router.get("/users/{id}", response_model=UserRead)
async def get(id: int, session: AsyncSession = Depends(get_session)):
    return await session.get(User, id)  # filtered through UserRead
```

### ❌ Global DB session

```python
# Bad: shared session = shared transaction state across requests
db_session = SessionLocal()  # ❌ module-level

# Good: one session per request via Depends(get_session)
```

## Mental Model

Think of a FastAPI handler as a thin shell: **parse → authorize → delegate → serialize**. Pydantic owns parse and serialize, `Depends` owns authorize, your service layer owns the rest. When a handler grows past twenty lines or reaches into the ORM, push logic down into a service and keep the router readable.