---
name: fastapi-pro
description: "FastAPI with Pydantic v2: lifespan, settings, Annotated dependencies, async SQLAlchemy 2.0 sessions, error handlers, background work, httpx tests. Use when building or reviewing a FastAPI service."
metadata:
  risk: safe
  source: multi-agent-pipeline
  date_added: "2026-09-25"
---

# FastAPI Pro

FastAPI makes the happy path short: type-hint a function and you get parsing,
validation and an OpenAPI document. The hard parts are the ones the tutorial
skips: where resources are created and closed, how a session maps to a
transaction, what happens when someone calls blocking code inside `async def`,
and how to test all of it without a real network.

Language-level guidance (typing, dataclasses, Pydantic basics) is in
`python-patterns`; schema and migrations in `database-patterns`; test strategy
in `testing-backend`.

## 1. Layout

```text
app/
  main.py            # create_app(): routers, middleware, handlers, lifespan
  settings.py        # pydantic-settings
  db.py              # engine, session factory, get_session dependency
  errors.py          # domain exceptions and handlers
  notes/
    router.py        # HTTP only: parse, call service, return schema
    schemas.py       # NoteCreate, NoteUpdate, NoteRead
    models.py        # SQLAlchemy ORM classes
    service.py       # business rules
    repository.py    # queries
tests/
  conftest.py
  test_notes.py
```

Keep routers thin. A router function that contains query logic and business
rules cannot be reused by a CLI command or a worker.

## 2. Core wiring

Load `references/app-structure.md` for the full code. The rules it implements:

- **Settings**: one `pydantic-settings` class read from the environment, cached
  behind `get_settings()` so tests can override it; the app fails to start on a
  missing or malformed value.
- **Lifespan**: `create_app()` builds routers, middleware and handlers; engines,
  HTTP clients and caches are created in the `lifespan` context and closed on
  shutdown, stored on `app.state`, never at import time.
- **Session per request**: a `get_session` dependency opens an `AsyncSession`
  from an `async_sessionmaker(expire_on_commit=False)` and wraps it in
  `session.begin()`, so the request commits on success and rolls back on any
  exception. Relationships are loaded explicitly (`selectinload`), never lazily.
- **Schemas**: separate input models (`NoteCreate`, `NoteUpdate` with
  `extra="forbid"`, no server-owned fields) from output models (`NoteRead`,
  `from_attributes=True`); endpoints declare `response_model`.
- **Dependencies**: `Annotated` aliases (`SessionDep`, `CurrentUser`) compose
  authentication, services and scoping; every query filters by the principal.
- **Errors**: services raise domain exceptions; one handler module maps them,
  and `RequestValidationError`, to RFC 9457 problem details.

## 3. async def versus def

- `async def` endpoints run on the event loop. Any blocking call inside them
  (a sync database driver, `requests`, `time.sleep`, heavy CPU work) stalls every
  concurrent request.
- Plain `def` endpoints run in a thread pool. Use them when the code is
  synchronous and you cannot change it.
- Offload blocking calls from async code with `await anyio.to_thread.run_sync(fn, ...)`;
  CPU-bound work belongs in a process pool or a job queue.
- Use async clients throughout: `asyncpg` or `psycopg` 3 async under SQLAlchemy,
  `httpx.AsyncClient`, `redis.asyncio`.

## 4. Background work

- `BackgroundTasks` runs after the response in the same process. It is fine for
  best-effort, short work (an audit log line). It is lost on crash or deploy.
- Anything that must happen (emails, webhooks, report generation, retries) goes
  to a durable queue: Celery, Dramatiq, arq, Taskiq, or a cloud queue. Make jobs
  idempotent and keyed.
- Enqueue in the same transaction as the domain change using an outbox table
  when losing the job would be a bug.

## 5. Testing and deployment

Tests create the app with `create_app()`, override dependencies such as
`get_current_user` and `get_settings`, and call it through `httpx.AsyncClient`
with `ASGITransport`. The transport does not run the lifespan, so wrap the app
in `asgi_lifespan.LifespanManager`. Fixtures, example tests and production
server settings (Uvicorn, proxy headers, instrumentation) are in
`references/testing-and-deployment.md`, together with notes on streaming
responses and WebSockets.

## Review checklist

- [ ] Resources are created in `lifespan` and closed on shutdown.
- [ ] One session per request, transaction committed or rolled back by the dependency.
- [ ] No blocking calls inside `async def`; relationships loaded explicitly.
- [ ] Input models forbid unknown and server-owned fields; responses use output models.
- [ ] Every query is scoped by the current principal.
- [ ] Domain exceptions map to problem details in one place.
- [ ] Durable work uses a queue, not `BackgroundTasks`.
- [ ] Tests use `AsyncClient` with lifespan and dependency overrides.

## References

- `references/app-structure.md`: settings, lifespan, session dependency, models, schemas, routers and error handlers in full.
- `references/testing-and-deployment.md`: test fixtures, dependency overrides, Uvicorn in production, streaming and WebSockets.

## Related skills

- `python-patterns`: typing, Pydantic, async fundamentals.
- `api-patterns`, `rest-api-design`: contract design.
- `api-security-best-practices`: authentication and authorization dependencies.
- `database-patterns`: schema, indexes, Alembic migrations.
- `testing-backend`: fixtures and test pyramid.