Design validated FastAPI endpoints with dependency injection, explicit errors, and testable boundaries.

Build typed APIs with FastAPI

FastAPI turns Python type hints and Pydantic models into request validation, response serialization, and OpenAPI documentation. Types improve the boundary, but the application still needs explicit domain rules, persistence boundaries, and error policy.

Create the project

uv add "fastapi[standard]"
uv add --dev pytest httpx

Use a package layout:

src/inventory/
├── __init__.py
├── api.py
├── models.py
├── repository.py
└── service.py
tests/
├── unit/
└── integration/

Define separate input and output models

# src/inventory/models.py
from typing import Annotated

from pydantic import BaseModel, ConfigDict, Field

NonBlankName = Annotated[str, Field(min_length=1, max_length=120)]


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

    name: NonBlankName
    quantity: Annotated[int, Field(ge=0)] = 0


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

    name: NonBlankName | None = None
    quantity: Annotated[int, Field(ge=0)] | None = None


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

    id: int
    name: str
    quantity: int

Separate models prevent clients from setting server-managed fields such as IDs. extra="forbid" catches misspelled fields instead of silently dropping them.

Depend on a repository protocol

# src/inventory/repository.py
from typing import Protocol

from inventory.models import Item, ItemCreate


class ItemRepository(Protocol):
    def list(self, *, offset: int, limit: int) -> list[Item]: ...

    def get(self, item_id: int) -> Item | None: ...

    def create(self, item: ItemCreate) -> Item: ...

    def close(self) -> None: ...

This boundary lets unit tests use an in-memory fake while integration tests exercise SQLite or PostgreSQL.

Build the application

# src/inventory/api.py
from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Query, status

from inventory.database import build_repository
from inventory.models import Item, ItemCreate
from inventory.repository import ItemRepository


def get_repository() -> Iterator[ItemRepository]:
    repository = build_repository()
    try:
        yield repository
    finally:
        repository.close()


Repository = Annotated[ItemRepository, Depends(get_repository)]

app = FastAPI(
    title="Inventory API",
    version="1.0.0",
    description="Manage inventory items.",
)


@app.get("/health", tags=["operations"])
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/items", response_model=list[Item], tags=["items"])
def list_items(
    repository: Repository,
    offset: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 50,
) -> list[Item]:
    return repository.list(offset=offset, limit=limit)


@app.get("/items/{item_id}", response_model=Item, tags=["items"])
def get_item(item_id: int, repository: Repository) -> Item:
    item = repository.get(item_id)
    if item is None:
        raise HTTPException(status_code=404, detail="item not found")
    return item


@app.post(
    "/items",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    tags=["items"],
)
def create_item(payload: ItemCreate, repository: Repository) -> Item:
    return repository.create(payload)

Implement build_repository() in the database adapter so it returns an ItemRepository. Keep construction in the composition layer rather than hiding connection details in route functions.

Run development mode:

uv run fastapi dev src/inventory/api.py --port 9000

Visit /docs for Swagger UI, /redoc for ReDoc, and /openapi.json for the machine-readable contract.

Sync versus async routes

Declare async def when the complete call path uses non-blocking async libraries:

@app.get("/remote-items/{item_id}")
async def get_remote_item(
    item_id: int,
    client: Annotated[AsyncCatalogClient, Depends(get_catalog_client)],
) -> Item:
    return await client.get_item(item_id)

A blocking database driver or Requests call inside async def blocks the event loop. Use a normal def route for blocking code, or select an async driver and await it throughout the stack. Do not mix models casually; measure under realistic concurrency.

Map domain errors centrally

from fastapi import Request
from fastapi.responses import JSONResponse


class DuplicateItemError(Exception):
    pass


@app.exception_handler(DuplicateItemError)
async def duplicate_item_handler(
    request: Request,
    exc: DuplicateItemError,
) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_409_CONFLICT,
        content={"detail": "an item with that name already exists"},
    )

Do not leak SQL errors, stack traces, or secrets. Log internal diagnostics with a request correlation ID and return a stable public error shape.

Test through ASGI

from fastapi.testclient import TestClient

from inventory.api import app, get_repository


class FakeRepository:
    def __init__(self) -> None:
        self.items: dict[int, Item] = {}

    def get(self, item_id: int) -> Item | None:
        return self.items.get(item_id)


def test_missing_item_returns_404() -> None:
    repository = FakeRepository()
    app.dependency_overrides[get_repository] = lambda: repository
    try:
        with TestClient(app) as client:
            response = client.get("/items/999")
    finally:
        app.dependency_overrides.clear()

    assert response.status_code == 404
    assert response.json() == {"detail": "item not found"}

Use TestClient as a context manager when lifespan events matter. Clear dependency overrides so tests cannot contaminate one another.

Production concerns

Configure allowed origins narrowly when browser clients require CORS.
Terminate TLS at a trusted proxy or platform and configure forwarded headers correctly.
Apply authentication and authorization separately; identity does not imply permission.
Bound request sizes, pagination limits, and upload sizes.
Make write retries safe with idempotency keys when duplicate effects matter.
Use structured logs and metrics, but exclude credentials and personal data.
A health endpoint should distinguish process liveness from dependency readiness when orchestration needs both.
Database migrations should run as a controlled deployment step, not concurrently in every worker.

Common pitfalls

Returning ORM objects without a deliberate response model.
Reusing one model for create, update, database, and public response.
Treating Pydantic validation as complete business validation.
Using async def around blocking I/O.
Creating HTTP or database clients on every route call.
Allowing unbounded list endpoints.
Returning 200 for creation when 201 and a resource location are more informative.

Practice

Add update and delete routes with correct 404, 409, and 204 behavior.
Reject unknown request fields and verify the generated OpenAPI schema.
Override a repository dependency in a route test.
Add a readiness endpoint that checks the database with a short timeout.
Design a stable error document with a code, message, and correlation ID.

Build typed APIs with FastAPI

Build typed APIs with FastAPI

Create the project

Define separate input and output models

Depend on a repository protocol

Build the application

Sync versus async routes

Map domain errors centrally

Test through ASGI

Production concerns

Common pitfalls

Practice

Finish this lesson

Build typed APIs with FastAPI

On this page