This is how I set up clean and maintainable **FastAPI** projects: DI, testing, validation, logging, deployment, auth, and more.
<!--more-->

> [!TIP]
> I encourage you to feed the contents of this cookbook into your AI agent of choice. Append `raw.md` to any page URL to get the raw markdown (e.g. [this page's source](raw.md) would give you the raw markdown of all the pages in this section (concatenated), or the "Source" icon next to any individual page's title).

## What's inside

Sections are ordered from foundational to advanced:

1. [Project Structure](project-structure) -- directory layout, `pyproject.toml` with `uv`, app factory pattern
2. [Configuration](configuration) -- `pydantic-settings`, nested settings with prefixes, environment-specific overrides
3. [Dependency Injection](dependency-injection) -- idiomatic DI with `Depends`, lifespan, `app.state`, and when to reach for `dependency-injector`
4. [Validation & Serialization](validation) -- Pydantic models, custom validators, `Annotated` field metadata
5. [Error Handling](error-handling) -- custom exceptions, problem-detail responses (RFC 9457), validation error reshaping
6. [Testing](testing) -- `pytest` + `httpx.AsyncClient`, `dependency_overrides`, fixtures
7. [Logging](logging) -- loguru setup, structured JSON output, streaming to CloudWatch
8. [Auth](auth) -- OAuth2 with JWT, security schemes via `Depends`, role-based access
9. [Deployment](deployment) -- multi-stage Dockerfile, `uvicorn` production settings, health checks


---

# Project Structure


How to lay out a FastAPI project so it stays maintainable as it grows.

<!--more-->

## Directory Layout

### Small projects (flat)

For a handful of endpoints with no database, a flat layout is fine:

```
my-api/
├── src/
│   ├── __init__.py
│   ├── app.py           # create_app factory, lifespan
│   ├── main.py          # entry point: app = create_app()
│   ├── config.py         # Settings (pydantic-settings)
│   ├── dependencies.py   # Depends helpers + type aliases
│   ├── schemas.py        # Pydantic request/response models
│   └── routes.py         # all route handlers
├── tests/
│   ├── __init__.py
│   └── test_routes.py
├── pyproject.toml
├── .env
└── .env.example
```

### Production projects (layered)

Once you have multiple resources, background tasks, or a database, split by layer:

```
my-api/
├── src/
│   ├── __init__.py
│   ├── app.py             # create_app factory, lifespan
│   ├── main.py            # entry point: app = create_app()
│   ├── config.py           # Settings
│   ├── dependencies.py     # shared Depends (settings, db session)
│   ├── exceptions.py       # custom exception classes
│   ├── middleware.py        # CORS, request-id, logging middleware (optional?)
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── users.py        # UserCreate, UserResponse, ...
│   │   └── bookings.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── users.py        # SQLAlchemy / ORM models
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── users.py        # router = APIRouter(prefix="/users")
│   │   └── bookings.py
│   ├── services/
│   │   ├── __init__.py
│   │   └── users.py        # business logic, orchestration
│   └── repositories/
│       ├── __init__.py
│       └── users.py        # raw DB queries
├── tests/
│   ├── __init__.py
│   ├── conftest.py           # shared fixtures (app, client, db)
│   ├── unit/
│   │   ├── __init__.py
│   │   ├── test_schemas.py   # pure validation, no I/O
│   │   └── test_services.py  # business logic with mocked repos
│   └── integration/
│       ├── __init__.py
│       ├── conftest.py       # real DB, test containers
│       ├── test_users.py     # full request → response
│       └── test_bookings.py
├── pyproject.toml
├── .env
└── Dockerfile
```

**What goes where:**

| Layer | Responsibility |
|---|---|
| `routers/` | HTTP concerns only: parse request, call service, return response |
| `services/` | Business logic, orchestration, validation that depends on state |
| `repositories/` | Data access --- SQL queries, external API calls |
| `schemas/` | Pydantic models for request/response (no DB coupling) |
| `models/` | ORM models (SQLAlchemy, etc.) |
| `dependencies.py` | `Depends` helpers and `Annotated` type aliases |
| `exceptions.py` | Custom exception classes (see [Error Handling](../error-handling)) |
| `tests/unit/` | Fast, no I/O --- schema validation, service logic with mocked deps |
| `tests/integration/` | Slower, real DB or test containers --- full request/response cycles |

> [!NOTE]
> Some teams prefer domain-driven grouping (everything for `users` in one folder). Both work --- the layered approach shown here is easier to follow when the team is small and the domain boundaries are still fuzzy.

## `pyproject.toml` with `uv`

```toml
[project]
name = "my-api"
version = "0.1.0"
requires-python = ">=3.13"
dependencies = [
    "fastapi[standard]",     # includes uvicorn + the fastapi CLI
    "pydantic-settings",
    "sqlalchemy[asyncio]",
]

[dependency-groups]
dev = [
    "ruff",
    "pre-commit",
]
test = [
    "pytest",
    "pytest-asyncio",
    "httpx",                 # for AsyncClient in tests
]

[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"

[tool.ruff]
target-version = "py313"
line-length = 88

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
```

Install everything with:

```bash
uv sync --all-groups
```

> [!NOTE]
> `fastapi[standard]` pulls in `uvicorn[standard]` (with `uvloop` + `httptools`), the `fastapi` CLI, and other production essentials. Always use this extra unless you have a reason to pin a specific `uvicorn` version.

## Running the App

The `fastapi` CLI (bundled with `fastapi[standard]`) provides two commands:

```bash
# Development: auto-reload on file changes, debug mode
uv run fastapi dev

# Production-like: no reload, single worker
uv run fastapi run
```

**Auto-discovery:** the CLI looks for a variable named `app` in `main.py` or `app.py` (in that order). If your app lives elsewhere, pass the path explicitly:

```bash
uv run fastapi dev src/main.py
```

| Command | Reload | Debug | Use case |
|---|---|---|---|
| `uv run fastapi dev` | yes | yes | Local development |
| `uv run fastapi run` | no | no | Containers, staging, CI |
| `uv run uvicorn src.main:app` | manual flags | manual flags | Full control over workers, host, port |

For production deployments behind a reverse proxy, you'll typically use `uvicorn` directly or a process manager like `gunicorn` with `uvicorn` workers --- see the [Deployment](../deployment) page.

## The Entry Point

For anything beyond a toy project, use a `create_app()` factory. It keeps initialization explicit and makes testing straightforward (each test can get a fresh app).

Keep the factory in `src/app.py` --- separate from `src/main.py` --- so that tests can `from src.app import create_app` without triggering the module-level `app = create_app()` as a side effect.

```python
# src/app.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from src.dependencies import get_settings
from src.exception_handlers import register_exception_handlers
from src.routers import bookings, users


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()
    # set up DB pools, caches, etc. on app.state
    # (see the Dependency Injection page for the full pattern)
    yield
    # teardown


def create_app() -> FastAPI:
    app = FastAPI(title="My API", lifespan=lifespan)

    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # ⚠ dev only — see warning below
        allow_methods=["*"],
        allow_headers=["*"],
    )

    register_exception_handlers(app)

    app.include_router(users.router)
    app.include_router(bookings.router)

    return app
```

```python
# src/main.py
from src.app import create_app

app = create_app()
```

`main.py` is the file the FastAPI CLI discovers (`fastapi dev` looks for an `app` variable in `main.py`). Tests import `create_app` from `src.app` and never touch `main.py`.

> [!WARNING]
> `allow_origins=["*"]` is fine for local development but **must not** be used in production --- it allows any website to make credentialed requests to your API. In production, list your specific origins (e.g. `["https://app.example.com"]`). See the [FastAPI CORS docs](https://fastapi.tiangolo.com/tutorial/cors/).

> [!WARNING]
> Don't put heavy initialization (DB connections, HTTP clients) inside `create_app()` directly --- use the lifespan context manager instead. `create_app()` should be fast and side-effect-free so tests can call it freely. See [Dependency Injection](../dependency-injection) for the lifespan pattern.

A typical router file looks like this:

```python
# src/routers/users.py
from fastapi import APIRouter

from src.dependencies import DbSession, ServiceDep
from src.schemas.users import UserCreate, UserResponse

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


@router.post("/", status_code=201)
async def create_user(body: UserCreate, db: DbSession) -> UserResponse:
    ...

@router.get("/")
async def get_users(db: DbSession) -> List[UserResponse]:
    ...

@router.get("/{user_id}")
async def get_user(user_id: int, db: DbSession) -> UserResponse:
    ...
```

## Configuration

See the [Configuration](../configuration) page for the full `Settings` class setup: `.env` loading, nested settings with prefixes, and environment-specific overrides. For how to inject `Settings` into route handlers via `Depends`, see [Dependency Injection](../dependency-injection#1-settings-read-only-config-from-env--env).


---

# Configuration


Managing application settings with `pydantic-settings` --- type-safe, validated, and environment-aware.

<!--more-->

## The Settings Class
```python
# src/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        extra="ignore",          # don't crash on unknown env vars
    )

    app_name: str = "My API"
    debug: bool = False
    database_url: str = "sqlite+aiosqlite:///dev.db"
    cors_origins: list[str] = ["*"]
```

Values are resolved in this order (first match wins):

1. Environment variables (case-insensitive)
2. `.env` file
3. Field defaults

```bash
# .env
DATABASE_URL=postgresql+asyncpg://user:pass@localhost/mydb
```

> [!NOTE]
> `extra="ignore"` is important in practice. Without it, any unknown environment variable that happens to match a prefix will cause a validation error at startup.

For how to inject `Settings` into route handlers via `Depends`, see the [Dependency Injection](../dependency-injection#1-settings-read-only-config-from-env--env) page.

## Nested Settings

As the settings class grows, group related config into nested `BaseModel` sub-models. Set `env_nested_delimiter` on the root `Settings` so pydantic-settings can map environment variables like `DB__POOL_SIZE` to `settings.db.pool_size`:

```python
# src/config.py
from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict


class DatabaseSettings(BaseModel):
    url: str = "sqlite+aiosqlite:///dev.db"
    password: SecretStr = SecretStr("")
    pool_size: int = 5
    echo: bool = False


class RedisSettings(BaseModel):
    url: str = "redis://localhost:6379/0"
    ttl: int = 3600


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_nested_delimiter="__",
        extra="ignore",
    )

    app_name: str = "My API"
    debug: bool = False
    cors_origins: list[str] = ["*"]
    db: DatabaseSettings = DatabaseSettings()
    redis: RedisSettings = RedisSettings()
```

```bash
# .env
DB__URL=postgresql+asyncpg://user:pass@localhost/mydb
DB__PASSWORD=s3cret
DB__POOL_SIZE=20
REDIS__URL=redis://cache:6379/0
```

The field name becomes the prefix automatically --- `db: DatabaseSettings` maps to `DB__*` env vars. Access in code reads naturally: `settings.db.url`, `settings.redis.ttl`.

> [!NOTE]
> Nested sub-models inherit from `BaseModel`, **not** `BaseSettings`. Only the root class should be `BaseSettings` --- this is the pattern the [pydantic-settings docs](https://docs.pydantic.dev/latest/concepts/pydantic_settings/) document for nested configuration. `env_prefix` on sub-models has no effect; the delimiter handles nesting.


## Why use `pydantic-settings` when we have...

### `dotenv`
`python-dotenv` loads `.env` into `os.environ` and gives you strings back. You still have to cast types, set defaults, and validate manually. `pydantic-settings` does all of that declaratively --- types, defaults, validation, and `.env` loading in one class.

### `os.getenv`
`os.getenv("DATABASE_URL", "sqlite:///dev.db")` works for one or two values. At ten settings it becomes a scattered mess of string casts, missing-key bugs, and no single place to see what the app expects. A `Settings` class is that single place, and you get type errors at startup instead of runtime `KeyError`s in production.

Also, with a `Settings` object you get two ways to control config in tests: swap the entire object via `dependency_overrides` (see [Dependency Injection](../dependency-injection#1-settings-read-only-config-from-env--env)), or patch individual env vars and let `Settings()` re-read them. Neither option exists when config is scattered across bare `os.getenv` calls.


## Practical tips

Use `SecretStr` for any field that holds a secret (passwords, API keys, tokens). It prevents the value from leaking into logs, tracebacks, and `model_dump()` output:

```python
from pydantic import SecretStr

password: SecretStr = SecretStr("")

settings.db.password.get_secret_value()  # access the actual value
str(settings.db.password)                # → '**********'
```


---

# Dependency Injection


Patterns for wiring settings, repositories, services, DB pools, and per-request objects into FastAPI route handlers --- using only what the framework gives you.

<!--more-->

> [!NOTE]
> This cookbook uses `Annotated` type aliases throughout (available since FastAPI 0.95+). Older tutorials show `param: Settings = Depends(get_settings)` --- that still works, but the `Annotated` form is cleaner and reusable across multiple endpoints.

## The Problem

You have objects (settings, repositories, services, DB pools) that route handlers need. You need to control **how many** instances exist, **when** they're created and destroyed, and make them **swappable in tests**.

FastAPI's native DI covers two lifetimes out of the box:

- **Per-process** singletons via the lifespan context manager
- **Per-request** objects via `Depends` (optionally with `yield` for teardown)

That handles the vast majority of real-world apps. The examples below use [SQLModel](https://sqlmodel.tiangolo.com/) (sync) for the SQL layer and [motor](https://motor.readthedocs.io/) for MongoDB. [Section 5](#5-with-async-sqlalchemy) shows how the same patterns look with async SQLAlchemy.

Here is how all the pieces fit together --- each numbered section below explains one part of this diagram:

```mermaid
graph TD
    subgraph lifespanPhase ["Lifespan (process startup)"]
        getSettings["get_settings()"]
        createEngine["create_engine()"]
        motorClient["AsyncIOMotorClient()"]
        auditRepo["AuditRepository()"]
        auditSvc["AuditService()"]

        getSettings --> createEngine
        getSettings --> motorClient
        motorClient --> auditRepo
        auditRepo --> auditSvc
    end

    createEngine -.->|"app.state.engine"| getSession
    auditSvc -.->|"app.state.audit_service"| getAuditSvc

    subgraph sqlChain ["Per-request DI chain (SQL)"]
        getSession["get_session()"]
        getRepo["get_booking_repo()"]
        getBookingSvc["get_booking_service()"]

        getSession -->|SessionDep| getRepo
        getRepo -->|BookingRepoDep| getBookingSvc
    end

    subgraph mongoSingleton ["Singleton (MongoDB)"]
        getAuditSvc["get_audit_service()"]
    end

    getBookingSvc -->|BookingServiceDep| handler["Route handler"]
    getAuditSvc -->|AuditServiceDep| handler
```

## 1. Settings (read-only config from env / `.env`)

**Pattern: `lru_cache` + `Depends`**

`Settings()` reads from disk (`.env`) or the environment, so you memoize it. Since settings are immutable after creation, `lru_cache` is safe. (For the full `Settings` class setup --- nested settings, prefixes, env-specific overrides --- see the [Configuration](../configuration) page.)

```python
# src/dependencies.py
from functools import lru_cache
from typing import Annotated

from fastapi import Depends

from src.config import Settings


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


# Reusable type alias --- use this in any handler signature
SettingsDep = Annotated[Settings, Depends(get_settings)]
```

```python
# src/api.py
@router.get("/health")
def health(settings: SettingsDep) -> dict:
    return {"status": "ok", "debug": settings.debug}
```

**Testing:** override the dependency --- `lru_cache` is bypassed entirely. Point it at a disposable Postgres container so your tests run against the same engine as production (never swap Postgres for SQLite --- query behaviour, types, and constraints differ). [testcontainers-python](https://testcontainers-python.readthedocs.io/) spins one up in a fixture. See [Testing](../testing#dependency_overrides) for the full fixture pattern.

```python
app.dependency_overrides[get_settings] = lambda: Settings(
    db=DatabaseSettings(url="postgresql://test:test@localhost:5433/test_db")
)
```

If you call `get_settings()` directly outside the FastAPI request cycle (e.g. in a CLI command or a test helper), `dependency_overrides` has no effect. Reset the cache instead: `get_settings.cache_clear()`.

## 2. Engine and connection pool (process-scoped)

**Pattern: Lifespan + `app.state`**

The database engine (and its underlying connection pool) is the one true process-scoped singleton in the SQL path. Create it in the lifespan context manager from settings and store it on `app.state` --- no module-level globals. Everything else (sessions, repos, services) is per-request (next section).

```python
# src/app.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from fastapi import FastAPI
from sqlmodel import create_engine

from src.dependencies import get_settings
from src.routers import bookings


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()
    engine = create_engine(
        str(settings.db.url),        # ← nested settings (see Configuration)
        pool_size=settings.db.pool_size,
        echo=settings.db.echo,
    )
    app.state.engine = engine
    yield
    engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(lifespan=lifespan)
    app.include_router(bookings.router)
    return app
```

Schema creation and migrations are handled by [Alembic](https://alembic.sqlalchemy.org/), not in the lifespan --- the lifespan is only for wiring runtime resources.

> [!NOTE]
> The [official full-stack-fastapi-template](https://github.com/fastapi/full-stack-fastapi-template) uses a module-level global `engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))`. That works fine --- the engine is a thread-safe connection pool --- but `app.state` integrates better with the app factory pattern and `dependency_overrides` in tests: you can create multiple app instances with different engines without monkey-patching module-level globals.

> [!WARNING]
> Starlette's `app.state` is dynamically typed --- `request.app.state.engine` will cause Mypy / Pyright errors because the attribute doesn't exist in the `State` class definition. Suppress with `# type: ignore[attr-defined]`, or use `cast` to keep strict linters happy:
>
> ```python
> from typing import cast
> from sqlalchemy import Engine
>
> engine = cast(Engine, request.app.state.engine)
> ```

## 3. Per-request DI chain (SQL layer)

**Pattern: `Depends` sub-dependencies**

When the ORM requires a session scoped to a unit of work (SQLAlchemy / SQLModel), the session, repository, and service should all be **per-request**. FastAPI's [sub-dependency](https://fastapi.tiangolo.com/tutorial/dependencies/sub-dependencies/) system wires them into a chain --- the "Per-request DI chain" box in the [diagram above](#the-problem). Each link is a `Depends` function. FastAPI resolves them right-to-left, caches each result within the request, and tears down `yield`-based dependencies after the response is sent.

```python
# src/dependencies.py
from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, Request
from sqlmodel import Session

from src.repositories.bookings import BookingRepository
from src.services.bookings import BookingService


def get_session(request: Request) -> Iterator[Session]:
    with Session(request.app.state.engine) as session:
        yield session


SessionDep = Annotated[Session, Depends(get_session)]


def get_booking_repo(session: SessionDep) -> BookingRepository:
    return BookingRepository(session)


BookingRepoDep = Annotated[BookingRepository, Depends(get_booking_repo)]


def get_booking_service(repo: BookingRepoDep) -> BookingService:
    return BookingService(repo)


BookingServiceDep = Annotated[BookingService, Depends(get_booking_service)]
```

The repo receives the session directly --- it never opens or closes sessions itself:

```python
# src/repositories/bookings.py
from collections.abc import Sequence

from sqlmodel import Session, select

from src.models.bookings import Booking


class BookingRepository:
    def __init__(self, session: Session) -> None:
        self.session = session

    def list_all(self) -> Sequence[Booking]:
        return self.session.exec(select(Booking)).all()

    def get_by_id(self, booking_id: int) -> Booking | None:
        return self.session.get(Booking, booking_id)
```

The service receives the repo --- it has no knowledge of sessions or the database:

```python
# src/services/bookings.py
from src.repositories.bookings import BookingRepository
from src.schemas.bookings import BookingResponse


class BookingService:
    def __init__(self, repo: BookingRepository) -> None:
        self.repo = repo

    def list_bookings(self) -> list[BookingResponse]:
        return [BookingResponse.model_validate(b) for b in self.repo.list_all()]
```

The endpoint declares only `BookingServiceDep`. FastAPI resolves the entire chain --- `get_booking_service` → `get_booking_repo` → `get_session` --- and tears down the session after the response:

```python
# src/routers/bookings.py
from fastapi import APIRouter

from src.dependencies import BookingServiceDep
from src.schemas.bookings import BookingResponse

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


@router.get("/")
def list_bookings(service: BookingServiceDep) -> list[BookingResponse]:
    return service.list_bookings()
```

**Testing:** override at whatever level you need. Each override short-circuits everything below it in the chain. See [Testing](../testing#dependency_overrides) for the full fixture pattern.

```python
# swap the whole service (unit-test the endpoint in isolation)
app.dependency_overrides[get_booking_service] = lambda: BookingService(mock_repo)
# swap only the session (integration-test with a real repo against a test DB)
app.dependency_overrides[get_session] = lambda: test_session
```

**Key properties:**

- Each request gets its own session, repo, and service --- no shared mutable state between requests.
- `Depends` results are cached within a request: if multiple dependencies need the same `SessionDep`, the session is created only once.
- Repos and services are lightweight objects --- constructing them per-request costs virtually nothing compared to the I/O they perform.

> [!TIP]
> **Why per-request, not singletons?** You could store the repo and service on `app.state` as singletons (passing the session factory to the repo and letting it manage sessions internally). That works, but:
>
> - The repo now manages session lifecycle (open/close per method), which is harder to test and reason about.
> - `get_repo` as a `Depends` becomes dead boilerplate --- endpoints only use services, so nobody calls it.
> - It contradicts the [FastAPI SQL tutorial](https://fastapi.tiangolo.com/tutorial/sql-databases/) and [SQLModel docs](https://sqlmodel.tiangolo.com/tutorial/fastapi/session-with-dependency/), which both inject sessions per-request.
>
> The singleton approach **is** correct for clients that manage their own connection pool and have no per-request session concept --- see the [next section](#4-singleton-services-non-sql-clients).

## 4. Singleton services (non-SQL clients)

**Pattern: Lifespan + `app.state` + `Depends`**

Not every data store has a per-request session concept. MongoDB (via `motor`), Redis, HTTP API clients, or compute-heavy objects like ML models all manage their own connection pool or have no teardown-per-request need. For these, the entire client → repo → service chain is process-scoped: create everything in the lifespan, attach **only the service** to `app.state`.

```python
# src/app.py (lifespan --- showing both SQL and MongoDB side by side)
from motor.motor_asyncio import AsyncIOMotorClient
from sqlmodel import create_engine

from src.repositories.audit import AuditRepository
from src.services.audit import AuditService


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()

    # SQL --- only the engine is process-scoped; sessions/repos/services are per-request
    engine = create_engine(str(settings.db.url))
    app.state.engine = engine

    # MongoDB --- full singleton chain (motor manages its own pool)
    mongo_client = AsyncIOMotorClient(str(settings.mongo.url))
    mongo_db = mongo_client[settings.mongo.database]
    audit_repo = AuditRepository(mongo_db)
    app.state.audit_service = AuditService(audit_repo)

    yield

    engine.dispose()
    mongo_client.close()
```

Only the service goes on `app.state`. The client and repo are internal wiring details --- endpoints never touch them, and exposing `get_audit_repo` as a `Depends` would be unnecessary boilerplate.

```python
# src/dependencies.py
def get_audit_service(request: Request) -> AuditService:
    return request.app.state.audit_service


AuditServiceDep = Annotated[AuditService, Depends(get_audit_service)]
```

**Testing:** override the one dependency:

```python
app.dependency_overrides[get_audit_service] = lambda: AuditService(mock_audit_repo)
```

**Mixed dependencies** --- a SQL-backed service can depend on both a per-request repo and a singleton:

```python
def get_booking_service(
    repo: BookingRepoDep, audit: AuditServiceDep
) -> BookingService:
    return BookingService(repo, audit)
```

FastAPI resolves both: `repo` is per-request (via the DI chain from [section 3](#3-per-request-di-chain-sql-layer)), `audit` is a singleton pointer-read from `app.state`.

**The rule of thumb:** if the client manages its own connection pool and has no request-scoped session, use singletons. If the ORM requires a session scoped to a unit of work (SQLAlchemy / SQLModel), use the per-request DI chain.

## 5. With async SQLAlchemy

The DI chain structure from [section 3](#3-per-request-di-chain-sql-layer) is identical when using async SQLAlchemy --- only the engine and session setup change. SQLModel table models work unchanged with `AsyncSession`.

```python
# src/app.py (lifespan)
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    settings = get_settings()
    engine = create_async_engine(str(settings.db.url))   # async driver, e.g. asyncpg
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()
```

```python
# src/dependencies.py
from collections.abc import AsyncIterator

from sqlalchemy.ext.asyncio import AsyncSession


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


SessionDep = Annotated[AsyncSession, Depends(get_session)]
```

The rest of the chain (`get_repo` → `get_service` → endpoint) stays the same --- swap `Session` for `AsyncSession` in the repo's type hints.

> [!NOTE]
> With async, you store a `session_factory` (an `async_sessionmaker`) on `app.state` rather than the raw engine, because `AsyncSession(engine)` is not the intended API --- `async_sessionmaker` configures session options (like `expire_on_commit`) in one place. With sync SQLModel, `Session(engine)` is the standard API, so storing the engine directly is fine.

## Decision Matrix

| What you're injecting | Pattern | Why |
|---|---|---|
| Read-only config (`Settings`) | `@lru_cache` + `Depends` | Docs-blessed; memoizes `.env` and environment read; no teardown needed |
| SQL engine / connection pool | Lifespan + `app.state` | Process-scoped; teardown via `engine.dispose()` after `yield` |
| SQL session, repo, service | `Depends` chain with sub-dependencies | ORM session must be scoped to a unit of work; per-request creation and teardown |
| Non-SQL client + repo + service (MongoDB, Redis, HTTP) | Lifespan + `app.state` + `Depends` | Client manages its own pool; no per-request session; only the service goes on `app.state` |

## Anti-Patterns

> [!WARNING]
> These are common mistakes that look convenient at first but cause real pain in testing, lifecycle management, or type safety.

| Anti-pattern | Why it's bad | Fix |
|---|---|---|
| Module-level `engine = create_engine(...)` | Couples to import-time settings; can't vary per test without monkey-patching | Create engine in lifespan from settings, store on `app.state` |
| Module-level `service = None  # type: ignore` set at startup | Not type-safe, not testable without monkey-patching, invisible lifecycle | Lifespan + `app.state` + `Depends` |
| `@lru_cache` on stateful dependencies (repos, services) | No teardown, outside FastAPI's lifecycle, awkward with `dependency_overrides` | Lifespan + `app.state` for singletons, `Depends` chain for per-request |
| Singleton repos that open/close sessions internally | Hides session scope; harder to test; risk of leaking sessions across requests | Per-request DI chain --- inject the session into the repo |
| Importing singletons directly (`from src.repo import repo`) | Untestable, hidden coupling, no lifecycle management | Always inject via `Depends` |

## Thread Safety

FastAPI runs **sync** route handlers in a threadpool. Shared objects may be accessed from multiple threads concurrently.

- **`Settings`** is safe --- it's read-only after creation.
- **Engine / connection pool** is safe --- SQLAlchemy's engine is designed for concurrent access.
- **Per-request repos and services** are safe --- each request gets its own instances, so there is no shared mutable state.
- **Singleton services** (non-SQL, on `app.state`) with mutable state need protection:
  - Sync handlers: use `threading.Lock`.
  - Async handlers: use `asyncio.Lock` --- concurrent coroutines interleave at `await` points.
  - Or delegate to a database / cache that handles concurrency internally.
- **Per-worker-process**: with `uvicorn --workers N` (or Gunicorn), each worker is a separate OS process with its own copy of `app.state`. Singletons are not shared across workers --- keep this in mind for in-memory caches, warmup data, and startup side effects.

## Further Reading

- [FastAPI - Dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/) --- how to use `Depends` to inject dependencies into route handlers.
- [FastAPI - Sub-dependencies](https://fastapi.tiangolo.com/tutorial/dependencies/sub-dependencies/) --- how to build dependency chains where one dependency depends on another.
- [FastAPI - SQL Databases](https://fastapi.tiangolo.com/tutorial/sql-databases/) --- official tutorial using SQLModel with `Depends` for session injection.
- [SQLModel - Session with FastAPI Dependency](https://sqlmodel.tiangolo.com/tutorial/fastapi/session-with-dependency/) --- SQLModel's recommended session-per-request pattern.
- [FastAPI - Lifespan Events](https://fastapi.tiangolo.com/advanced/events/) --- how to use the lifespan context manager to create and manage resources.
- [full-stack-fastapi-template](https://github.com/fastapi/full-stack-fastapi-template) --- official project template showing SQLModel + FastAPI in production.
- [Starlette - State](https://www.starlette.io/applications/#storing-state-on-the-app-instance) --- how to store state on the app instance.
- [dependency-injector](https://python-dependency-injector.ets-labs.org/) --- third-party DI container; useful when you need DI outside route handlers (Celery tasks, CLI commands) or have deep dependency graphs. Check out how I use it here in my [secret santa telegram bot](https://github.com/rayannott/ded-moroz/blob/main/src/dependencies.py).


---

# Validation & Serialization


Using Pydantic to validate incoming data and shape outgoing responses.

<!--more-->

## Schemas: Input vs Output

A single model for both input and output leaks internal fields (password hashes, internal IDs) and accepts fields the client shouldn't control (`id`, `created_at`). Split into separate schemas.

```python
# src/schemas/users.py
from datetime import datetime

from pydantic import BaseModel, EmailStr


class UserBase(BaseModel):
    """Fields shared between input and output."""
    email: EmailStr
    display_name: str


class UserCreate(UserBase):
    """What the client sends when creating a user."""
    password: str


class UserUpdate(BaseModel):
    """Partial update --- all fields optional."""
    email: EmailStr | None = None
    display_name: str | None = None


class UserResponse(UserBase):
    """What the API returns. No password, has server-generated fields."""
    id: int
    created_at: datetime
```

Use them in route handlers:

```python
# src/routers/users.py
@router.post("/", status_code=201)
async def create_user(body: UserCreate, db: DbSession) -> UserResponse:
    user = await user_service.create(db, body)
    return UserResponse.model_validate(user)
```

> [!NOTE]
> `UserUpdate` does not inherit from `UserBase` because partial updates need every field to be optional. A shared base with required fields would defeat the purpose.

## Field Constraints with `Annotated`

Use `Annotated` with `Field`, `Query`, `Path`, `Body`, and `Header` to declare constraints once and reuse them across endpoints.

### Body fields (`Field`)

```python
from typing import Annotated

from pydantic import BaseModel, Field


class BookingCreate(BaseModel):
    room_id: Annotated[int, Field(ge=1, description="ID of the room to book")]
    guests: Annotated[int, Field(ge=1, le=20)]
    note: Annotated[str, Field(max_length=500)] = ""
```

### Query and path parameters

```python
from typing import Annotated

from fastapi import Path, Query


@router.get("/rooms")
async def list_rooms(
    floor: Annotated[int, Query(ge=1, le=50, description="Filter by floor")] = 1,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> list[RoomResponse]:
    ...


@router.get("/rooms/{room_id}")
async def get_room(
    room_id: Annotated[int, Path(ge=1)],
) -> RoomResponse:
    ...
```

### Headers

```python
from fastapi import Header


@router.get("/protected")
async def protected(
    x_api_key: Annotated[str, Header()],
) -> dict:
    ...
```

### Reusable constraints

Define constraint types once and use them everywhere:

```python
# src/schemas/common.py
from typing import Annotated

from pydantic import Field

PositiveInt = Annotated[int, Field(ge=1)]
ShortStr = Annotated[str, Field(min_length=1, max_length=255)]
```

```python
# src/schemas/bookings.py
from src.schemas.common import PositiveInt, ShortStr


class BookingCreate(BaseModel):
    room_id: PositiveInt
    title: ShortStr
```

## Custom Validators

### Single-field validation (`@field_validator`)

```python
from pydantic import BaseModel, field_validator


class SignupRequest(BaseModel):
    username: str
    email: str

    @field_validator("username")
    @classmethod
    def username_must_be_alphanumeric(cls, v: str) -> str:
        if not v.isalnum():
            raise ValueError("must be alphanumeric")
        return v.lower()  # normalize to lowercase
```

By default, `@field_validator` runs **after** Pydantic's built-in type coercion (mode `"after"`). Use `mode="before"` to transform raw input before type checking:

```python
@field_validator("tags", mode="before")
@classmethod
def split_tags(cls, v: str | list[str]) -> list[str]:
    if isinstance(v, str):
        return [t.strip() for t in v.split(",")]
    return v
```

### Cross-field validation (`@model_validator`)

```python
from pydantic import BaseModel, model_validator


class DateRange(BaseModel):
    start: datetime
    end: datetime

    @model_validator(mode="after")
    def end_must_be_after_start(self) -> "DateRange":
        if self.end <= self.start:
            raise ValueError("end must be after start")
        return self
```

`mode="after"` gives you a fully constructed model instance --- all fields are already validated and typed. Use `mode="before"` when you need to inspect or reshape raw input data before individual field validation runs.

### Reusable validators with `Annotated`

For validation logic you reuse across models, attach validators directly to the type:

```python
from typing import Annotated

from pydantic import AfterValidator


def must_be_lowercase(v: str) -> str:
    if v != v.lower():
        raise ValueError("must be lowercase")
    return v


LowercaseStr = Annotated[str, AfterValidator(must_be_lowercase)]
```

Now any field typed as `LowercaseStr` gets the validation automatically --- no decorator needed on every model.

## Serialization Control

### Return type vs `response_model`

Modern FastAPI (0.95+) infers the response schema from the return type annotation. When the return type *is* the public schema, this is the cleanest approach:

```python
# Return type is the public schema --- no response_model needed
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: DbSession) -> UserResponse:
    ...

# Use response_model when the API contract differs from the
# implementation return type (filtering, coercion, union disambiguation)
@router.get("/users/me", response_model=UserPublic)
async def get_current_user(db: DbSession) -> User:
    ...
```

When both are present, `response_model` takes priority.

### ORM integration (`from_attributes`)

To construct a Pydantic model directly from a SQLAlchemy (or any ORM) instance:

```python
from pydantic import BaseModel, ConfigDict


class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    display_name: str
```

This lets you pass an ORM object directly:

```python
user = await db.get(User, user_id)
return UserResponse.model_validate(user)  # reads from user.id, user.email, etc.
```

### Computed fields

For fields derived from other data at serialization time:

```python
from pydantic import BaseModel, computed_field


class BookingResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    room_id: int
    guests: int
    price_per_guest: float

    @computed_field
    @property
    def total_price(self) -> float:
        return self.guests * self.price_per_guest
```

`total_price` appears in the JSON response and the OpenAPI schema, but is never stored --- it's computed on every serialization.


---

# Error Handling


Turning exceptions into consistent, machine-readable API responses.

<!--more-->

## The Defaults

FastAPI ships with `HTTPException` for quick error responses:

```python
from fastapi import HTTPException


@router.get("/users/{user_id}")
async def get_user(user_id: int, db: DbSession) -> UserResponse:
    user = await db.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return UserResponse.model_validate(user)
```

This returns `{"detail": "User not found"}` with a 404 status code.

> [!NOTE]
> `detail` accepts any JSON-serializable value, not just strings. You can pass a dict or list for structured errors:
> ```python
> raise HTTPException(
>     status_code=400,
>     detail={"code": "INVALID_STATE", "message": "Booking is already confirmed"},
> )
> ```

`HTTPException` works fine for simple APIs. But as your app grows, you'll want:

- A consistent error shape across all endpoints
- Error types that carry domain semantics (`NotFoundError`, not `HTTPException(404)`)
- Machine-readable error codes, not just human messages

## Custom Exceptions

Define a base exception that carries everything needed to build an error response:

```python
# src/exceptions.py
class AppError(Exception):
    """Base exception for all application errors."""

    def __init__(
        self,
        status_code: int = 500,
        title: str = "Internal Server Error",
        detail: str = "",
    ) -> None:
        self.status_code = status_code
        self.title = title
        self.detail = detail


class NotFoundError(AppError):
    def __init__(self, resource: str, resource_id: object) -> None:
        super().__init__(
            status_code=404,
            title="Not Found",
            detail=f"{resource} {resource_id} not found",
        )


class ConflictError(AppError):
    def __init__(self, detail: str = "Resource already exists") -> None:
        super().__init__(status_code=409, title="Conflict", detail=detail)


class ForbiddenError(AppError):
    def __init__(self, detail: str = "Insufficient permissions") -> None:
        super().__init__(status_code=403, title="Forbidden", detail=detail)
```

Now route handlers read like domain code:

```python
@router.get("/users/{user_id}")
async def get_user(user_id: int, db: DbSession) -> UserResponse:
    user = await db.get(User, user_id)
    if not user:
        raise NotFoundError("User", user_id)
    return UserResponse.model_validate(user)
```

## Exception Handlers

FastAPI needs a handler to convert `AppError` into an HTTP response. Keep handlers in their own module to avoid cluttering `create_app`:

```python
# src/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from src.exceptions import AppError


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
        return JSONResponse(
            status_code=exc.status_code,
            content={"title": exc.title, "detail": exc.detail},
        )
```

```python
# src/app.py
from src.exception_handlers import register_exception_handlers


def create_app() -> FastAPI:
    app = FastAPI(...)
    register_exception_handlers(app)
    # ... routers, middleware ...
    return app
```

Any `AppError` (or subclass) raised anywhere --- in route handlers, dependencies, or services --- is now caught and serialized consistently.

## Problem-Detail Responses (RFC 9457)

[RFC 9457](https://www.rfc-editor.org/rfc/rfc9457) defines a standard JSON shape for API errors. Adopting it means every client, SDK generator, and monitoring tool can parse your errors without custom logic.

A problem-detail response looks like this:

```json
{
    "type": "https://my-api.example.com/errors/not-found",
    "title": "Not Found",
    "status": 404,
    "detail": "User 42 not found",
    "instance": "/users/42"
}
```

You don't need a library for this --- it's a Pydantic model and an exception handler:

```python
# src/schemas/errors.py
from pydantic import BaseModel


class ProblemDetail(BaseModel):
    type: str = "about:blank"
    title: str
    status: int
    detail: str = ""
    instance: str = ""
```

```python
# src/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from src.exceptions import AppError
from src.schemas.errors import ProblemDetail

ERROR_TYPE_BASE = "https://my-api.example.com/errors"


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(request: Request, exc: AppError) -> JSONResponse:
        slug = exc.title.lower().replace(" ", "-")
        body = ProblemDetail(
            type=f"{ERROR_TYPE_BASE}/{slug}",
            title=exc.title,
            status=exc.status_code,
            detail=exc.detail,
            instance=str(request.url.path),
        )
        return JSONResponse(
            status_code=exc.status_code,
            content=body.model_dump(exclude_none=True),
            media_type="application/problem+json",
        )
```

> [!NOTE]
> Setting `media_type="application/problem+json"` is part of the RFC spec. It tells clients the body follows the problem-detail format. Most JSON parsers handle it transparently.

This replaces the simpler handler from the previous section --- the `AppError` hierarchy stays exactly the same.

> [!WARNING]
> The `type` URI derived from `exc.title` is a quick start, but RFC 9457 expects type URIs to be **stable identifiers**. If a title changes, the URI changes, breaking clients that match on it. For production APIs, define type URIs as explicit constants on each exception class (e.g. `error_type = "not-found"`) instead of deriving them from presentation text.

## Reshaping Validation Errors

FastAPI's default response for invalid request data (`RequestValidationError`) looks like this:

```json
{
    "detail": [
        {
            "type": "missing",
            "loc": ["body", "email"],
            "msg": "Field required",
            "input": {},
            "url": "https://errors.pydantic.dev/2.11/v/missing"
        }
    ]
}
```

The `url` field (added by Pydantic v2) is noisy for API consumers, and the shape doesn't match the problem-detail format. Add a handler to `register_exception_handlers` with a second handler for `RequestValidationError`:

```python
# src/exception_handlers.py  (continued)
from fastapi.exceptions import RequestValidationError


def register_exception_handlers(app: FastAPI) -> None:
    @app.exception_handler(AppError)
    async def app_error_handler(...) -> JSONResponse: ...  # as above

    @app.exception_handler(RequestValidationError)
    async def validation_error_handler(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        errors = []
        for err in exc.errors():
            errors.append({
                "field": ".".join(str(loc) for loc in err["loc"]),
                "message": err["msg"],
                "type": err["type"],
            })

        body = ProblemDetail(
            type=f"{ERROR_TYPE_BASE}/validation-error",
            title="Validation Error",
            status=422,
            detail=f"{len(errors)} validation error(s)",
            instance=str(request.url.path),
        )
        payload = body.model_dump()
        payload["errors"] = errors

        return JSONResponse(
            status_code=422,
            content=payload,
            media_type="application/problem+json",
        )
```

The response now looks like this:

```json
{
    "type": "https://my-api.example.com/errors/validation-error",
    "title": "Validation Error",
    "status": 422,
    "detail": "2 validation error(s)",
    "instance": "/users/",
    "errors": [
        {"field": "body.email", "message": "Field required", "type": "missing"},
        {"field": "body.password", "message": "Field required", "type": "missing"}
    ]
}
```

> [!WARNING]
> The `errors` extension field is allowed by RFC 9457 (the spec is extensible), but not all tools will know about it. The top-level `title`, `status`, and `detail` fields are always enough for a human or a generic error handler to understand what went wrong.

> [!WARNING]
> Reshaping `RequestValidationError` changes the response contract. Clients and SDK generators that expect FastAPI's native validation shape (`{"detail": [...]}`) will break. Decide early whether your API uses the default shape or problem-detail --- switching later is a breaking change.


---

# Testing


Testing FastAPI apps with `pytest`, `httpx.AsyncClient`, and `pytest-asyncio`. For project setup (`pyproject.toml`, test directory layout, `asyncio_mode`), see [Project Structure](../project-structure#pyprojecttoml-with-uv).

<!--more-->

## The Test Client

Use `httpx.AsyncClient` with `ASGITransport` to test your app in-process --- no real network, native async, and the same API you'd use in production code:

```python
@pytest.fixture
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
```

`base_url` can be anything --- requests never leave the process. The `app` parameter is a fixture that builds a fresh `FastAPI` instance per test (shown in the next section).

> [!NOTE]
> Starlette's sync `TestClient` works, but `AsyncClient` is the better default for FastAPI: it supports `async def` endpoints natively and mirrors the `httpx` API you'd use to call external services.

## `dependency_overrides`

FastAPI's DI and its testing story are two sides of the same coin. The [Dependency Injection](../dependency-injection) page shows the production wiring --- `get_settings`, `get_booking_service`, `get_session` --- each used in `Depends()`. In tests, you swap them via `app.dependency_overrides`.

It's a dict mapping a `Depends` callable to its replacement. FastAPI checks it **before** resolving any dependency, short-circuiting the entire chain:

```
Production:  handler → get_booking_service → get_booking_repo → get_session → app.state.engine

Test:        handler → dependency_overrides[get_booking_service] → lambda: test_service
```

You don't need to mock `app.state`, the lifespan, or any intermediate step --- override the `get_*` function and FastAPI does the rest. The entire engine → session → repo → service chain is bypassed. This is why the [DI patterns](../dependency-injection) use thin `get_*` wrappers rather than inline logic: each wrapper is a clean seam you can swap in tests.

### The fixture pattern

Wire overrides in the `app` fixture. Each mock (settings, repo) comes in as its own fixture parameter, so pieces stay independently swappable. The `app` fixture calls `create_app()`, applies overrides, `yield`s, then calls `app.dependency_overrides.clear()` in teardown to prevent leakage between tests.

You can override at any level of the [dependency chain](../dependency-injection) --- `get_settings` to swap config, `get_booking_repo` to replace just the repository, `get_session` to point at a test database, or `get_booking_service` to bypass the entire chain. The complete `conftest.py` is in the [next section](#fixtures).

> [!WARNING]
> The override key must be the **exact callable** used in `Depends()`. If the route uses `Depends(get_booking_service)`, the key is `get_booking_service` --- not the class, not a string, not a different wrapper.

## Fixtures

Complete `conftest.py` tying everything together:

```python
# tests/conftest.py
from collections.abc import AsyncIterator
from unittest.mock import MagicMock

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient

from src.app import create_app
from src.config import Settings
from src.dependencies import get_settings, get_booking_service
from src.repositories.bookings import BookingRepository
from src.services.bookings import BookingService


@pytest.fixture
def test_settings() -> Settings:
    return Settings(debug=True)


@pytest.fixture
def mock_repo() -> MagicMock:
    return MagicMock(spec=BookingRepository)


@pytest.fixture
async def app(
    test_settings: Settings,
    mock_repo: MagicMock,
) -> AsyncIterator[FastAPI]:
    app = create_app()
    app.dependency_overrides[get_settings] = lambda: test_settings
    app.dependency_overrides[get_booking_service] = lambda: BookingService(mock_repo)
    yield app
    app.dependency_overrides.clear()


@pytest.fixture
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
```

All fixtures default to `function` scope --- each test gets a fresh app, fresh overrides, and a fresh client. Widen scope only when test speed demands it and you've verified there's no shared mutable state.

> [!NOTE]
> `AsyncClient` with `ASGITransport` does **not** trigger [lifespan events](https://fastapi.tiangolo.com/advanced/async-tests/#httpx). That's fine here --- `dependency_overrides` bypasses everything the lifespan would set up, so `app.state` never needs to be populated. For integration tests that use real dependencies (no overrides), wrap the app in [`LifespanManager`](https://github.com/florimondmanca/asgi-lifespan) so the lifespan runs:
>
> ```python
> from asgi_lifespan import LifespanManager
>
> @pytest.fixture
> async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
>     async with LifespanManager(app):
>         async with AsyncClient(
>             transport=ASGITransport(app=app),
>             base_url="http://test",
>         ) as ac:
>             yield ac
> ```

## Integration tests (real database)

Unit tests with mocked services are fast but don't catch SQL bugs, schema mismatches, or driver-specific behaviour. Integration tests hit a real database --- **the same engine as production**. Never swap Postgres for SQLite --- query semantics, types, and constraints differ.

Use [testcontainers-python](https://testcontainers-python.readthedocs.io/) to spin up a disposable Postgres container. The container starts once per session; each test gets its own transaction that rolls back on teardown:

```python
# tests/integration/conftest.py
from collections.abc import AsyncIterator, Iterator

import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlmodel import Session, SQLModel, create_engine
from testcontainers.postgres import PostgresContainer

from src.app import create_app
from src.dependencies import get_session


@pytest.fixture(scope="session")
def engine():
    with PostgresContainer("postgres:16") as pg:
        engine = create_engine(pg.get_connection_url())
        SQLModel.metadata.create_all(engine)
        yield engine
        engine.dispose()


@pytest.fixture
def session(engine) -> Iterator[Session]:
    with Session(engine) as session:
        yield session
        session.rollback()


@pytest.fixture
async def app(session: Session) -> AsyncIterator[FastAPI]:
    app = create_app()
    app.dependency_overrides[get_session] = lambda: session
    yield app
    app.dependency_overrides.clear()


@pytest.fixture
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    async with AsyncClient(
        transport=ASGITransport(app=app),
        base_url="http://test",
    ) as ac:
        yield ac
```

Only `get_session` is overridden --- the real repo and service are constructed per-request as normal, exercising the full [DI chain](../dependency-injection#3-per-request-di-chain-sql-layer) against a live Postgres instance. See [Project Structure](../project-structure) for the `tests/integration/` directory layout.

> [!NOTE]
> **Testcontainers trade-offs:** Docker must be available --- both locally and in CI. The first container start adds a few seconds of overhead (session-scoping amortizes this across all tests). Testcontainers also starts a [Ryuk](https://github.com/testcontainers/moby-ryuk) sidecar to garbage-collect orphaned containers; some locked-down CI environments block this (set `TESTCONTAINERS_RYUK_DISABLED=true` as a workaround, but you risk leaked containers on crashes). On GitHub Actions with `ubuntu-latest`, Docker is pre-installed natively (the runner is a full VM, not a container) --- no extra configuration needed. However, if your CI runner is itself a container (e.g. GitHub Actions with `container:`, GitLab CI, or Kubernetes-based runners), you need [Docker-in-Docker](https://jpetazzo.github.io/2015/09/03/do-not-use-docker-in-docker-for-ci/): either mount the host Docker socket or run a DinD sidecar service.

### Coverage

Add [`pytest-cov`](https://pytest-cov.readthedocs.io/) to your test dependencies and run:

```bash
uv run pytest --cov=src --cov-report=term-missing
```

### Example

```python
# tests/test_rooms.py
from httpx import AsyncClient


async def test_list_rooms(client: AsyncClient) -> None:
    response = await client.get("/rooms")

    assert response.status_code == 200
    assert isinstance(response.json(), list)


async def test_health_returns_debug_flag(client: AsyncClient) -> None:
    response = await client.get("/health")

    assert response.status_code == 200
    assert response.json()["debug"] is True
```

## Further Reading

- [FastAPI --- Testing](https://fastapi.tiangolo.com/tutorial/testing/)
- [httpx --- Async Client](https://www.python-httpx.org/async/)
- [pytest-asyncio](https://pytest-asyncio.readthedocs.io/)
- [testcontainers-python](https://testcontainers-python.readthedocs.io/) --- disposable Docker containers for integration tests (Postgres, Redis, etc.)
- [polyfactory](https://polyfactory.litestar.dev/) --- auto-generate test data from Pydantic models
- [hypothesis](https://hypothesis.readthedocs.io/) --- property-based testing
- [pytest-cov](https://pytest-cov.readthedocs.io/)


---

# Logging


Setting up logging for a FastAPI application (or any other Python application, for that matter).

<!--more-->

## Why loguru

Python's stdlib `logging` works, but the configuration is verbose: handlers, formatters, log levels per logger, `basicConfig` vs `dictConfig`. `loguru` replaces all of that with a single `logger` object that's ready to use on import.

```bash
uv add loguru
```

```python
from loguru import logger

logger.info("Server started")
logger.warning("Slow query | elapsed={elapsed:.1f}s", elapsed=1.2)
logger.error("Failed to connect to DB")
```
> [!NOTE]
> Here, `loguru` interpolates the `elapsed` variable into the message string. This also lays the groundwork for structured logging, where the message is a template and the variables are the data.

That's it --- no `getLogger`, no handler setup, no formatter strings. The default output is colorized, includes timestamps, levels, module/function/line, and looks good out of the box.

## Log Level via Environment

loguru reads the `LOGURU_LEVEL` environment variable to set the minimum log level. No code changes needed --- just set it in your `.env` or deployment config:

```bash
# .env
LOGURU_LEVEL=DEBUG    # dev
```
Speaking of local development, [direnv](https://direnv.net/) can auto-load `.env` files when you `cd` into the project directory. Decrease the mental overhead of having to remember these small details and rather spend your brain power on the more important things.

```bash
# production
LOGURU_LEVEL=INFO
```

This means the log level lives alongside your other config (see [Configuration](../configuration)) without needing to wire it through `Settings`.


## Exception Tracebacks

`logger.exception()` catches the current exception and prints a full traceback alongside the log message --- useful during development:

```python
try:
    result = await fetch_user(user_id)
except Exception:
    logger.exception("Failed to fetch user {user_id}", user_id=user_id)
```

loguru's tracebacks are more readable than stdlib's: they include variable values at each frame, colorized output, and the full chain for re-raised exceptions. You can also use `logger.opt(exception=True).error(...)` for the same effect when you want a level other than `ERROR`.

In production, full tracebacks add noise and can leak internals. Disable them by setting `diagnose=False` on the handler:

```python
def setup_logging(json_logs: bool = False) -> None:
    logger.remove()

    logger.add(
        sys.stderr,
        serialize=json_logs,
        backtrace=True,   # show full call chain
        diagnose=False,    # show variable values in frames only in dev
    )
```

With `diagnose=False`, tracebacks are still logged (via `logger.exception`), but they look like standard Python tracebacks without the extra variable introspection. With `backtrace=False`, only the frames from the exception would be shown, not the full call chain leading up to the `try` block --- personally, I prefer to see the full call chain: for my own sanity, it's either no traceback at all, or the full traceback.

## Structured JSON Output

The default formatter is great for local development. For production, switch to JSON so your log aggregator (CloudWatch, Datadog, etc.) can parse fields automatically:

```python
import sys

from loguru import logger


def setup_logging(json_logs: bool = False) -> None:
    logger.remove()  # remove the default stderr handler

    if json_logs:
        logger.add(sys.stderr, serialize=True)  # JSON lines
    else:
        logger.add(sys.stderr)                   # pretty-printed default
```

`serialize=True` outputs each log entry as a JSON object with `text`, `record.level`, `record.time`, `record.message`, `record.extra`, etc.

Toggle via your settings:

```python
# src/main.py
setup_logging(json_logs=not settings.debug)
```

## Streaming to CloudWatch

With JSON logging enabled, CloudWatch (or any log aggregator that reads stdout/stderr) picks up the structured output automatically --- no special integration needed. The container runtime streams stderr to the log driver.

For ECS / Fargate, the default `awslogs` log driver captures everything written to stdout/stderr and sends it to CloudWatch Logs. Just make sure:

1. JSON logging is on (`serialize=True`)
2. The task role has `logs:CreateLogStream` and `logs:PutLogEvents` permissions
3. The log group exists (create it in your IaC, don't rely on auto-creation)

No loguru plugin or CloudWatch SDK is needed --- the log driver handles transport. I'll speak more about this in the [Deployment](../deployment) page.

> [!TIP]
> For correlating logs across services (trace IDs, span IDs, distributed tracing), see [Observability / Distributed Tracing](../../observability/distributed-tracing).


---

# Auth


Authentication and authorization patterns for FastAPI.

<!--more-->

## OAuth2 with JWT

## Security Schemes via `Depends`

## Role-Based Access

## Testing Authenticated Endpoints


---

# Deployment


Getting a FastAPI app into production.

<!--more-->

## Multi-Stage Dockerfile

## `uvicorn` Production Settings

## Health Checks

## Graceful Shutdown