FastAPI Services skill for Claude Code

A sound FastAPI service never blocks the event loop, declares request and response shapes as Pydantic models so the OpenAPI schema is accurate, gets every resource through dependencies that tests can override, and opens shared clients once in the lifespan.

Workflow

- [ ] 1. Check versions: python, fastapi, pydantic (must be v2), sqlalchemy, server
- [ ] 2. Map app layout: routers, dependencies, models vs schemas, DB session handling
- [ ] 3. Decide sync vs async per endpoint based on the libraries it calls
- [ ] 4. Define Pydantic v2 schemas (input, output, update) separately from ORM models
- [ ] 5. Wire resources via lifespan + Annotated dependencies
- [ ] 6. Handle errors consistently; move slow work out of the request
- [ ] 7. Test with dependency overrides; check the generated OpenAPI

1. Versions

python -c "import fastapi, pydantic, sys; print(sys.version, fastapi.__version__, pydantic.VERSION)"

Pydantic v1 idioms that must change: class Config to model_config = ConfigDict(...), .dict() to .model_dump(), .parse_obj() to .model_validate(), @validator to @field_validator, orm_mode to from_attributes=True, BaseSettings moved to the pydantic-settings package. @app.on_event("startup") is deprecated; use lifespan.

2. Layout

app/
  main.py          # create_app(), lifespan, include_router
  config.py        # Settings(BaseSettings)
  db.py            # engine, async_sessionmaker, get_session dependency
  deps.py          # Annotated aliases: SessionDep, CurrentUser, SettingsDep
  routers/orders.py
  schemas/orders.py   # Pydantic: OrderCreate, OrderRead, OrderUpdate
  models/orders.py    # SQLAlchemy ORM
  services/orders.py  # business logic, takes a session, no Request objects

3. Sync vs async: the decision that breaks most services

Endpoint calls	Declare as	Why
Only async libraries (httpx.AsyncClient, asyncpg, SQLAlchemy async)	async def	Runs on the event loop
Any blocking library (requests, psycopg2, boto3, sync SQLAlchemy, heavy CPU)	plain def	FastAPI runs it in a threadpool
Mostly async plus one blocking call	async def + await run_in_threadpool(fn, ...)	Isolates the blocking part
CPU-heavy work (image processing, ML inference > 100 ms)	Offload to a worker/process pool or queue	Threads do not escape the GIL

One time.sleep or requests.get inside async def stalls every concurrent request on that worker. Search for it: grep -rn "requests\.\|time\.sleep\|boto3" app/routers.

4. Schemas

from datetime import datetime
from decimal import Decimal
from typing import Annotated
from pydantic import BaseModel, ConfigDict, Field

class OrderCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")          # reject unknown fields
    sku: Annotated[str, Field(min_length=1, max_length=64)]
    quantity: Annotated[int, Field(ge=1, le=1000)]

class OrderRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)    # build from ORM objects
    id: int
    sku: str
    quantity: int
    total: Decimal
    created_at: datetime

class OrderUpdate(BaseModel):
    quantity: Annotated[int | None, Field(ge=1, le=1000)] = None

Apply partial updates with patch.model_dump(exclude_unset=True) so omitted fields are not overwritten with None. Always set response_model (or a return annotation) so internal fields like password_hash cannot leak.

5. Lifespan, settings, sessions

from contextlib import asynccontextmanager
from typing import Annotated, AsyncIterator
import httpx
from fastapi import Depends, FastAPI
from pydantic_settings import BaseSettings, SettingsConfigDict
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    database_url: str
    debug: bool = False

settings = Settings()
engine = create_async_engine(settings.database_url, pool_size=10, max_overflow=5, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    app.state.http = httpx.AsyncClient(timeout=10)
    yield
    await app.state.http.aclose()
    await engine.dispose()

async def get_session() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

app = FastAPI(lifespan=lifespan)

@app.post("/orders", response_model=OrderRead, status_code=201)
async def create_order(body: OrderCreate, session: SessionDep, user: CurrentUser):
    order = await orders_service.create(session, user, body)
    await session.commit()
    return order

Async SQLAlchemy traps: lazy-loaded relationships raise MissingGreenlet in async code; load what the response needs with selectinload()/joinedload() in the query, or use lazy="raise" on relationships to catch it in tests. expire_on_commit=False avoids reloads after commit.

6. Errors and slow work

• Raise HTTPException(status_code=404, detail="Order not found") for HTTP-level errors; map domain exceptions in one place with @app.exception_handler(DomainError).
• Customize RequestValidationError handling only to change the shape (e.g. RFC 9457); keep 422 as the status.
• BackgroundTasks runs after the response in the same process: fine for sending an email or logging, lost on crash or deploy. Anything that must happen, retries, or takes more than a few seconds goes to a real queue (Celery, arq, Dramatiq, SQS).

7. Tests

import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app
from app.db import get_session

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

async def test_create_order_rejects_zero_quantity(client):
    r = await client.post("/orders", json={"sku": "A1", "quantity": 0})
    assert r.status_code == 422

With pytest-asyncio set asyncio_mode = "auto" in pyproject.toml. ASGITransport does not run lifespan events; use asgi-lifespan’s LifespanManager or the sync with TestClient(app) as c: when the test needs them.

Running

fastapi dev app/main.py                      # reload, dev only
fastapi run app/main.py --workers 4          # production entry (uvicorn underneath)
uvicorn app.main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips="*"

Workers per container: start at the CPU count; DB pool size times workers times replicas must stay under the database connection limit.

Gotchas

• Dependencies with yield run cleanup after the response is sent; do not rely on them to change the response.
• Mutable default arguments in models are copied by Pydantic, but Depends() results are cached per request: two dependencies asking for get_session share one session.
• Returning ORM objects without from_attributes=True fails validation at runtime.
• Optional[X] without a default is still required in Pydantic v2; add = None.
• Path operations are matched in declaration order: declare /orders/export before /orders/{order_id}.

Verification

1. pytest passes, including validation and auth failure cases.
2. Open /docs or fetch /openapi.json: every route shows request and response schemas; no internal fields appear.
3. Load-test one async endpoint with 50 concurrent requests while another endpoint is slow; latency of the first should not rise (event loop not blocked).
4. ruff check and mypy or pyright pass on app/.