FastAPI Services skill for Codex

Done well means: routes are thin (parse, authorize, call a service, return a response model), shared resources are created once in lifespan, nothing blocking runs inside async def, every endpoint declares a response model that excludes internal fields, and tests replace dependencies instead of patching modules.

Inspect first

• Read AGENTS.md and pyproject.toml: FastAPI and Pydantic versions (v2 assumed), package manager (uv, poetry, pip), DB stack (SQLAlchemy async, SQLModel, asyncpg directly), test runner.
• Find the app and routers: rg -n "FastAPI\(|APIRouter\(|include_router" ..
• Find dependencies and settings: rg -n "Depends\(|BaseSettings|lifespan" ..
• Check how it runs: fastapi run, uvicorn ..., gunicorn config, Dockerfile.

Layout

app/
  main.py            # create_app(), lifespan, include_router, exception handlers
  config.py          # Settings (pydantic-settings)
  db.py              # engine, async_sessionmaker, get_session dependency
  deps.py            # SessionDep, CurrentUser, SettingsDep (Annotated aliases)
  orders/
    router.py        # HTTP only
    schemas.py       # Pydantic models: OrderCreate, OrderRead
    service.py       # business logic, takes a session and plain values
    models.py        # SQLAlchemy ORM models
tests/

Settings and lifespan

# config.py
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")
    database_url: str
    jwt_secret: str
    cors_origins: list[str] = []

@lru_cache
def get_settings() -> Settings:
    return Settings()

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    settings = get_settings()
    app.state.engine = create_async_engine(settings.database_url, pool_pre_ping=True)
    app.state.sessionmaker = async_sessionmaker(app.state.engine, expire_on_commit=False)
    yield
    await app.state.engine.dispose()

def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    app.include_router(orders_router, prefix="/orders", tags=["orders"])
    register_exception_handlers(app)
    return app

app = create_app()

@app.on_event("startup") is deprecated; use lifespan. Missing settings now fail at boot instead of at the first request.

Dependencies

# db.py / deps.py
from typing import Annotated, AsyncIterator
from fastapi import Depends, HTTPException, Request, status
from sqlalchemy.ext.asyncio import AsyncSession

async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
    async with request.app.state.sessionmaker() as session:
        yield session

SessionDep = Annotated[AsyncSession, Depends(get_session)]

async def get_current_user(session: SessionDep, token: Annotated[str, Depends(oauth2_scheme)]) -> User:
    user = await authenticate(session, token)
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Invalid token",
                            headers={"WWW-Authenticate": "Bearer"})
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]

• Dependencies are cached per request: get_session used by both the route and get_current_user yields one session.
• Router-wide auth: APIRouter(dependencies=[Depends(get_current_user)]).
• Code after yield in a dependency runs after the response is sent by default (it supports streaming responses). Do not hand that session to BackgroundTasks; open a fresh session inside the task.

Routes and models

# orders/schemas.py
from pydantic import BaseModel, ConfigDict, Field

class OrderCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    sku: str = Field(min_length=1, max_length=64)
    qty: int = Field(gt=0, le=1000)

class OrderRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    sku: str
    qty: int
    status: str

# orders/router.py
from fastapi import APIRouter, HTTPException, status

router = APIRouter()

@router.post("", response_model=OrderRead, status_code=status.HTTP_201_CREATED)
async def create_order(body: OrderCreate, session: SessionDep, user: CurrentUser):
    order = await service.create_order(session, user_id=user.id, data=body)
    await session.commit()
    return order

@router.get("/{order_id}", response_model=OrderRead)
async def get_order(order_id: int, session: SessionDep, user: CurrentUser):
    order = await service.get_for_user(session, order_id, user.id)
    if order is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, "Order not found")
    return order

• Always set response_model (or a return annotation). Returning an ORM object without one serializes whatever is reachable, including fields you did not mean to expose.
• Separate input and output models; never reuse the ORM model or one model with optional password for both directions.
• Scope queries by owner in the service (WHERE user_id = :uid) and return 404, not 403, for other users’ objects unless existence is public.

async def or def

Endpoint does	Declare as
Awaits async libraries (asyncpg, httpx.AsyncClient, aioboto3)	async def
Calls blocking libraries (requests, psycopg2, boto3, heavy CPU)	def (runs in the threadpool)
Mix	async def + await run_in_threadpool(fn, ...) or asyncio.to_thread for the blocking part

One time.sleep or requests.get inside async def stalls every request on that worker. CPU-heavy work belongs in a job queue: see $background-jobs.

Async SQLAlchemy rules

• expire_on_commit=False, or attribute access after commit() triggers a lazy load and raises MissingGreenlet.
• Load relationships explicitly: select(Order).options(selectinload(Order.items)). Setting lazy="raise" on relationships turns accidental lazy loads into clear errors.
• One AsyncSession per request; never share a session across concurrent tasks (asyncio.gather over the same session fails).
• Migrations with Alembic (alembic revision --autogenerate -m ..., review the script, alembic upgrade head). For deeper ORM questions use $sqlalchemy-2.

Errors

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

def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(DomainError)
    async def domain_error(_: Request, exc: DomainError):
        return JSONResponse({"error": {"code": exc.code, "message": exc.message}}, exc.status_code)

Services raise domain errors; routers or handlers map them to HTTP. Validation errors return 422 with Pydantic’s error list; customize via a RequestValidationError handler only if clients need a different shape.

Testing

import pytest
from httpx import ASGITransport, AsyncClient

@pytest.fixture
async def client(session):
    app = create_app()
    app.dependency_overrides[get_session] = lambda: session
    app.dependency_overrides[get_current_user] = lambda: User(id=1, email="[email protected]")
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

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

ASGITransport does not run lifespan; create resources in fixtures or wrap the app with asgi-lifespan’s LifespanManager if the test needs them. Use pytest-asyncio or anyio markers as the project already does. Run the DB in a transaction rolled back per test, or use testcontainers for Postgres.

Running in production

• fastapi run app/main.py --workers 4 or uvicorn app.main:app --workers 4. Behind a proxy, enable proxy headers and restrict --forwarded-allow-ips to the proxy.
• Pool size x workers x replicas must fit the database’s connection limit.
• Disable docs in production only if policy requires it: FastAPI(docs_url=None, redoc_url=None).

Verification

uv run pytest -q            # or the command in AGENTS.md
uv run ruff check . && uv run mypy app    # if configured
uv run fastapi dev app/main.py           # then open /docs and exercise new routes

Check the generated OpenAPI (/openapi.json) shows the intended response model with no internal fields, and that an unauthenticated call to a protected route returns 401.

Gotchas

• Returning a Response/JSONResponse directly bypasses response_model filtering and validation; return data and let FastAPI serialize it unless you need custom headers.
• BackgroundTasks run in the same worker after the response; a crash loses them. Use a real queue for anything that must happen.
• Depends() on a class instantiates it per request; expensive clients belong in lifespan.
• Path order matters: declare /orders/export before /orders/{order_id}.

Output template

## FastAPI change: <summary>
- Endpoints: <method path -> response model, status codes>
- Dependencies: <added/changed>
- Models: <input/output schemas>
- DB: <queries, eager loads, migration id>
- Tests: <cases, overrides used>