FastAPI - Type-Driven APIs with Automatic Validation and Docs

Reading time: ~35 minutes | Level: Intermediate → Engineering

Before reading further, consider this puzzle:

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.post("/items")
async def create_item(item: Item):
    return item

# POST /items {"name": "widget", "price": 9.99}    → ?
# POST /items {"name": "widget", "price": "not_a_number"} → ?
# POST /items {"name": "widget"}                    → ?
# GET  /docs                                        → ?

The first request returns {"name": "widget", "price": 9.99}. The second returns a 422 Unprocessable Entity with a structured JSON body that tells the client exactly which field failed and why. The third also returns a 422 explaining that price is required. The fourth serves a full interactive Swagger UI - auto-generated, always up to date, requiring zero manual maintenance.

None of this required you to write a single line of validation code, error formatting, or documentation markup. FastAPI derived all of it from the Python type hints on your model and handler. This is not magic - it is the deliberate engineering of three libraries working in concert: Starlette, Pydantic, and Python's typing module.

What You Will Learn

• FastAPI's architecture: Starlette + Pydantic + type hints - how they compose
• ASGI vs WSGI: why async matters for I/O-bound services
• The full FastAPI request lifecycle as a sequence diagram
• Path, query, and body parameters - all type-annotated and validated
• Annotated + Query/Path/Body for field-level constraints
• Dependency injection with Depends() - database sessions, auth, shared logic
• Resource lifecycle management with Depends + yield
• Background tasks: fire-and-forget after the response is sent
• Response models: enforcing output shape, stripping sensitive fields
• Status codes, JSONResponse, FileResponse, StreamingResponse
• Exception handlers and custom exception classes
• Middleware: @app.middleware("http") and add_middleware
• OpenAPI: automatic /docs and /redoc, customisation
• Testing with TestClient and httpx.AsyncClient
• Router organisation with APIRouter

Prerequisites

• Lesson 03 (Flask) - contrasting Flask's synchronous model helps explain FastAPI's design choices
• Lesson 01 (HTTP Deep Dive) - HTTP methods, status codes, and headers underpin every endpoint
• Lesson 08 (Pydantic) - FastAPI uses Pydantic for all validation; understanding models is essential

Part 1 - Architecture: Three Libraries, One Framework

FastAPI is not a web server. It is a framework built on top of three independent components:

Your Code (type-annotated handlers)
         ↓
  Pydantic v2  ←  validation, serialisation, JSON schema generation
         ↓
  Starlette    ←  ASGI toolkit: routing, middleware, request/response objects
         ↓
  Uvicorn      ←  ASGI server: runs the event loop, accepts TCP connections

Understanding this layering explains FastAPI's behaviour precisely:

• Pydantic reads your type annotations on request models and generates JSON Schema. When a request arrives, FastAPI calls Pydantic to validate the parsed body against the schema. If validation fails, Pydantic raises a ValidationError; FastAPI catches it and returns a 422 with the structured error details.
• Starlette provides Request, Response, routing (APIRouter is a subclass of Starlette's Router), middleware support, static files, WebSocket support, and background tasks.
• Uvicorn (or Hypercorn, or Daphne) is the ASGI server - the process that binds to a TCP port, accepts HTTP connections, and calls your FastAPI application object via the ASGI interface.

FastAPI itself is the glue: it reads Python type annotations to generate the OpenAPI schema, constructs the dependency injection graph, formats ValidationError into 422 responses, and wires Starlette's routing to Pydantic-validated handlers.

Part 2 - ASGI vs WSGI

Understanding why FastAPI uses ASGI requires understanding what WSGI cannot do.

WSGI (Web Server Gateway Interface, PEP 3333) was designed for synchronous Python web frameworks - Django, Flask, Gunicorn. The interface is a callable:

def application(environ: dict, start_response: callable) -> Iterable[bytes]:
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"Hello"]

The problem is structural: WSGI is synchronous. When your handler calls await db.query(...), WSGI cannot yield control to handle another request. The thread blocks. To handle N concurrent I/O-bound requests you need N threads. Threads are expensive (each costs ~8 MB of stack memory on Linux).

ASGI (Asynchronous Server Gateway Interface) replaces the synchronous callable with an async one:

async def application(scope: dict, receive: callable, send: callable) -> None:
    # scope: connection metadata (type, path, headers, ...)
    # receive: coroutine to read incoming data (body chunks, websocket messages)
    # send: coroutine to send outgoing data
    await send({"type": "http.response.start", "status": 200, "headers": []})
    await send({"type": "http.response.body", "body": b"Hello"})

With ASGI + asyncio, a single OS thread can handle thousands of concurrent requests as long as I/O operations use await. While one request is waiting for a database query, the event loop runs other coroutines.

Aspect	WSGI	ASGI
Concurrency model	Thread-per-request	Single event loop + coroutines
I/O handling	Blocks thread	Yields to event loop
Concurrent requests (1 process)	Limited by thread count (typ. 4–8)	Thousands (limited by file descriptors)
WebSocket support	No	Yes (native)
Suitable for CPU-bound work	Yes (threads handle it)	No - CPU blocks the event loop
Examples	Flask + Gunicorn, Django + Gunicorn	FastAPI + Uvicorn, Starlette + Uvicorn

note

FastAPI runs def (non-async) endpoint functions in a thread pool automatically. If your handler is defined with def instead of async def, FastAPI offloads it to asyncio.run_in_executor so it does not block the event loop. You pay a small thread pool overhead, but it is safe. Only async def handlers run directly in the event loop.

Part 3 - The FastAPI Request Lifecycle

When validation fails at the Pydantic step, FastAPI raises RequestValidationError and an exception handler converts it to a 422 Unprocessable Entity response with a body like:

{
  "detail": [
    {
      "type": "float_parsing",
      "loc": ["body", "price"],
      "msg": "Input should be a valid number, unable to parse string as a number",
      "input": "not_a_number"
    }
  ]
}

The loc array tells the client exactly where (body → price field) the validation failed.

Part 4 - Parameters: Path, Query, Body

FastAPI derives parameter types from function signatures. The location of a parameter (path, query, or body) is inferred from:

• If the name appears in the path template → path parameter
• If the type is a Pydantic BaseModel → request body
• Otherwise → query parameter

from fastapi import FastAPI, Query, Path, Body
from pydantic import BaseModel
from typing import Annotated

app = FastAPI()

class ItemBody(BaseModel):
    name: str
    price: float
    description: str | None = None

@app.get("/items/{item_id}")
async def get_item(
    # Path parameter - inferred from {item_id} in route
    item_id: Annotated[int, Path(ge=1, description="The item ID, must be >= 1")],
    # Query parameters - everything else that is not a BaseModel
    q: Annotated[str | None, Query(min_length=1, max_length=50)] = None,
    include_deleted: bool = False,
) -> dict:
    return {"item_id": item_id, "q": q, "include_deleted": include_deleted}

@app.post("/items")
async def create_item(
    # Request body - Pydantic model
    item: ItemBody,
    # Query parameter alongside a body - explicit with Annotated
    dry_run: Annotated[bool, Query(description="If true, validate but do not save")] = False,
) -> ItemBody:
    return item

Constraint Reference

from fastapi import Query, Path, Body
from typing import Annotated

# Numeric constraints
page: Annotated[int, Query(ge=1, le=1000)] = 1          # 1 ≤ page ≤ 1000
price: Annotated[float, Query(gt=0.0)] = ...            # strictly positive

# String constraints
name: Annotated[str, Query(min_length=1, max_length=100)] = ...
slug: Annotated[str, Query(pattern=r"^[a-z0-9-]+$")] = ...

# Path parameter with description and alias
item_id: Annotated[int, Path(ge=1, title="Item ID", description="Primary key")]

# Body with example
item: Annotated[ItemBody, Body(
    openapi_examples={
        "widget": {
            "summary": "A typical item",
            "value": {"name": "widget", "price": 9.99}
        }
    }
)]

tip

Use Annotated[Type, Query(...)] rather than Query(...) as a default value. The Annotated pattern is cleaner: the type and the validation metadata are bundled together, and the actual default is expressed separately - making function signatures easier to read and type checkers happier.

Part 5 - Dependency Injection with Depends

Dependency injection in FastAPI is a function graph. A dependency is any callable that FastAPI knows to call before your handler. Dependencies can depend on other dependencies - FastAPI resolves the full directed acyclic graph.

from fastapi import FastAPI, Depends, HTTPException, status
from typing import Annotated

app = FastAPI()

# --- Dependency 1: parse and validate common pagination params ---
def pagination(
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 20,
) -> dict:
    return {"skip": skip, "limit": limit}

# --- Dependency 2: extract and validate auth token ---
def get_current_user(
    authorization: str = Header(default=None),
) -> dict:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing or invalid Authorization header",
        )
    token = authorization.removeprefix("Bearer ")
    # In real code: verify JWT, look up user
    return {"user_id": 42, "token": token}

# --- Dependency 3: depends on Dependency 2 ---
def require_admin(
    current_user: Annotated[dict, Depends(get_current_user)],
) -> dict:
    if current_user.get("role") != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Admin access required",
        )
    return current_user

# Handler uses both dependencies
@app.get("/items")
async def list_items(
    pagination: Annotated[dict, Depends(pagination)],
    user: Annotated[dict, Depends(get_current_user)],
) -> dict:
    # pagination and user are already validated by the time we get here
    return {"user_id": user["user_id"], **pagination}

@app.delete("/items/{item_id}")
async def delete_item(
    item_id: int,
    admin: Annotated[dict, Depends(require_admin)],  # transitively calls get_current_user
) -> dict:
    return {"deleted": item_id}

Dependencies can also be classes - FastAPI calls __init__ and uses the instance:

class CommonQueryParams:
    def __init__(
        self,
        skip: Annotated[int, Query(ge=0)] = 0,
        limit: Annotated[int, Query(ge=1, le=100)] = 20,
        sort_by: str = "created_at",
    ):
        self.skip = skip
        self.limit = limit
        self.sort_by = sort_by

@app.get("/products")
async def list_products(
    params: Annotated[CommonQueryParams, Depends()],
) -> dict:
    # params.skip, params.limit, params.sort_by
    return {"skip": params.skip, "limit": params.limit}

Depends with yield - Resource Lifecycle

For resources that need explicit cleanup (database sessions, file handles, locks), use yield in a dependency:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from typing import AsyncGenerator

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with SessionLocal() as session:
        try:
            yield session          # handler runs HERE with an open session
            await session.commit() # commit if handler completed without exception
        except Exception:
            await session.rollback()
            raise

@app.post("/users")
async def create_user(
    user_data: UserCreate,
    db: Annotated[AsyncSession, Depends(get_db)],
) -> UserResponse:
    user = User(**user_data.model_dump())
    db.add(user)
    # commit happens in the dependency's finally logic
    return UserResponse.model_validate(user)

Everything before yield runs before the handler. Everything after yield runs after - even if the handler raises an exception. This is equivalent to a context manager wrapping the handler execution.

Part 6 - Background Tasks

BackgroundTasks runs a function after the response has been sent to the client. The client does not wait for the background work.

from fastapi import FastAPI, BackgroundTasks
import smtplib
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

def send_welcome_email(email: str, username: str) -> None:
    # This runs AFTER the response is sent - client doesn't wait
    logger.info("Sending welcome email to %s", email)
    # ... smtp logic ...

def write_audit_log(action: str, user_id: int) -> None:
    # Write to a slow audit log without blocking the response
    logger.info("AUDIT: %s by user %d", action, user_id)

@app.post("/users", status_code=201)
async def register_user(
    user_data: UserCreate,
    background_tasks: BackgroundTasks,
    db: Annotated[AsyncSession, Depends(get_db)],
) -> UserResponse:
    user = await create_user_in_db(db, user_data)

    # Schedule background tasks - they run after this function returns
    background_tasks.add_task(send_welcome_email, user.email, user.username)
    background_tasks.add_task(write_audit_log, "user_registered", user.id)

    return UserResponse.model_validate(user)

warning

BackgroundTasks runs in the same process as the ASGI server - not in a separate worker. If the background function is CPU-bound or takes many seconds, it will degrade the event loop's performance. For heavy background work (sending bulk emails, processing files, running ML inference), use a proper task queue (Celery, ARQ, Dramatiq) with separate worker processes.

Part 7 - Response Models

The response_model parameter on an endpoint decorator does three things:

1. Filters the return value - only fields declared in the response model are included in the response (sensitive fields like password_hash are automatically stripped)
2. Validates the return value - raises an error during development if your handler returns a shape that doesn't match the declared model
3. Documents the response shape in the OpenAPI schema

from pydantic import BaseModel, EmailStr
from datetime import datetime

class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str  # raw password from client

class UserInDB(BaseModel):
    id: int
    username: str
    email: EmailStr
    password_hash: str  # sensitive - NEVER send to client
    created_at: datetime

class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr
    created_at: datetime
    # No password_hash - safe to return to client

@app.post(
    "/users",
    response_model=UserResponse,  # enforces output shape
    status_code=201,
    summary="Register a new user",
    description="Creates a user account and returns the public profile.",
    tags=["users"],
)
async def create_user(user_data: UserCreate) -> UserInDB:
    # Handler can return a UserInDB (which has password_hash)
    # FastAPI applies response_model filtering before serialisation
    # password_hash is automatically excluded from the response
    user_in_db = await save_user(user_data)
    return user_in_db  # safe: response_model strips password_hash

danger

Never return SQLAlchemy ORM objects directly from a handler. ORM objects are lazy-loaded - accessing their attributes outside a database session raises DetachedInstanceError. Always convert to a Pydantic response model before returning. Use UserResponse.model_validate(orm_obj) with model_config = ConfigDict(from_attributes=True) on your response model.

Response model options

# Exclude fields with None values from the response
@app.get("/items/{id}", response_model=ItemResponse, response_model_exclude_none=True)

# Include only specific fields
@app.get("/items/{id}", response_model=ItemResponse, response_model_include={"id", "name"})

# Exclude specific fields
@app.get("/items/{id}", response_model=ItemResponse, response_model_exclude={"internal_notes"})

Part 8 - Status Codes and Response Classes

from fastapi import FastAPI, Response, status
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
import io

app = FastAPI()

# Declarative status code
@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemBody) -> ItemBody:
    return item

# Return 204 No Content (no body)
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int) -> None:
    await remove_item(item_id)
    # Return None - FastAPI sends 204 with no body

# Dynamic status code via Response object
@app.put("/items/{item_id}")
async def upsert_item(item_id: int, item: ItemBody, response: Response) -> ItemBody:
    existing = await get_item(item_id)
    if existing:
        await update_item(item_id, item)
        response.status_code = status.HTTP_200_OK
    else:
        await insert_item(item_id, item)
        response.status_code = status.HTTP_201_CREATED
    return item

# JSONResponse for custom headers or cookies
@app.post("/login")
async def login(credentials: LoginRequest) -> JSONResponse:
    token = await authenticate(credentials)
    return JSONResponse(
        content={"access_token": token},
        headers={"X-Token-Issued": "true"},
    )

# FileResponse for downloads
@app.get("/report.pdf")
async def download_report() -> FileResponse:
    return FileResponse("reports/latest.pdf", media_type="application/pdf", filename="report.pdf")

# StreamingResponse for large or generated data
@app.get("/stream")
async def stream_data() -> StreamingResponse:
    async def generate():
        for i in range(1000):
            yield f"data: {i}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Part 9 - Exception Handlers

FastAPI provides two built-in exceptions: HTTPException for expected errors (404, 401, 403) and RequestValidationError for Pydantic validation failures.

from fastapi import FastAPI, HTTPException, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

# Custom exception class
class ItemNotFoundError(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# Register handler for your custom exception
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError) -> JSONResponse:
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={"error": "item_not_found", "item_id": exc.item_id},
    )

# Override the default 422 validation error format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    errors = [
        {"field": " → ".join(str(loc) for loc in err["loc"]), "message": err["msg"]}
        for err in exc.errors()
    ]
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={"errors": errors},
    )

@app.get("/items/{item_id}")
async def get_item(item_id: int) -> dict:
    item = await fetch_item(item_id)
    if not item:
        raise ItemNotFoundError(item_id=item_id)  # custom exception
    return item

Part 10 - Middleware

Middleware wraps every request and response. There are two ways to add middleware in FastAPI:

from fastapi import FastAPI, Request
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.gzip import GZipMiddleware
import time
import uuid

app = FastAPI()

# --- Method 1: decorator syntax ---
@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start) * 1000
    print(
        f"method={request.method} path={request.url.path} "
        f"status={response.status_code} duration_ms={duration_ms:.1f}"
    )
    return response

# --- Method 2: add_middleware (for class-based middleware) ---
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.add_middleware(GZipMiddleware, minimum_size=1000)

warning

async def in a FastAPI handler does not automatically make I/O non-blocking. If you call time.sleep(1) inside an async def handler, it blocks the event loop for 1 second - no other requests are processed. Use await asyncio.sleep(1), await db.query(...), await httpx_client.get(...). Any blocking I/O inside async def starves every concurrent request. If you cannot make something async, use def and FastAPI will run it in a thread pool.

Part 11 - OpenAPI: Automatic /docs and /redoc

FastAPI builds an OpenAPI 3.1 schema by introspecting your type annotations, response models, Depends() graph, and decorator parameters. The schema is served at /openapi.json, the Swagger UI at /docs, and ReDoc at /redoc.

from fastapi import FastAPI

app = FastAPI(
    title="Inventory API",
    description="Manages warehouse inventory with full audit trail.",
    version="2.3.1",
    contact={"name": "Platform Team", "email": "[email protected]"},
    license_info={"name": "MIT"},
    # Restrict which environments expose docs
    docs_url="/docs" if settings.ENV != "production" else None,
    redoc_url="/redoc" if settings.ENV != "production" else None,
)

@app.get(
    "/items/{item_id}",
    summary="Retrieve a single item",
    description="Returns the item with the given ID. Returns 404 if not found.",
    response_description="The item record",
    tags=["inventory"],
    deprecated=False,
    responses={
        404: {"description": "Item not found"},
        200: {"description": "Item found"},
    },
)
async def get_item(item_id: int) -> ItemResponse:
    ...

tip

Use response_model on every endpoint. It does three things simultaneously: enforces what your handler actually returns (catches bugs), strips fields you did not intend to expose, and documents the exact response schema in the generated OpenAPI spec - giving clients accurate auto-generated client SDKs.

Part 12 - Testing: TestClient and AsyncClient

# tests/test_items.py
import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport

from myapp.main import app
from myapp.database import get_db

# --- Synchronous testing with TestClient ---
@pytest.fixture
def client():
    with TestClient(app) as c:
        yield c

def test_create_item_valid(client):
    response = client.post("/items", json={"name": "widget", "price": 9.99})
    assert response.status_code == 201
    data = response.json()
    assert data["name"] == "widget"
    assert data["price"] == 9.99

def test_create_item_invalid_price(client):
    response = client.post("/items", json={"name": "widget", "price": "not_a_number"})
    assert response.status_code == 422
    errors = response.json()["detail"]
    assert any(err["loc"][-1] == "price" for err in errors)

def test_create_item_missing_field(client):
    response = client.post("/items", json={"name": "widget"})  # price missing
    assert response.status_code == 422

# --- Override dependencies for testing ---
def override_get_db():
    # Return a test database session instead of production
    yield test_db_session

app.dependency_overrides[get_db] = override_get_db

# --- Async testing with httpx ---
@pytest.mark.asyncio
async def test_create_item_async():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        response = await ac.post("/items", json={"name": "widget", "price": 9.99})
    assert response.status_code == 201

The TestClient wraps the ASGI app in a synchronous interface using requests-style calls. AsyncClient from httpx drives the ASGI app asynchronously - required when testing endpoints that use async database calls or when you need to test concurrent requests.

Part 13 - Router Organisation with APIRouter

As an API grows, all routes in a single file become unmanageable. APIRouter mirrors FastAPI's main app interface and lets you mount groups of routes with shared prefixes, tags, and dependencies.

myapp/
  main.py
  routers/
    __init__.py
    users.py
    items.py
    orders.py
  dependencies.py
  models.py

# myapp/routers/users.py
from fastapi import APIRouter, Depends, status
from myapp.dependencies import get_db, require_auth

router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(require_auth)],  # applies to ALL routes in this router
    responses={401: {"description": "Not authenticated"}},
)

@router.get("/")
async def list_users(db = Depends(get_db)) -> list[UserResponse]:
    ...

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

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_user(user: UserCreate, db = Depends(get_db)) -> UserResponse:
    ...

# myapp/main.py
from fastapi import FastAPI
from myapp.routers import users, items, orders

app = FastAPI(title="My API", version="1.0.0")

app.include_router(users.router)
app.include_router(items.router)
app.include_router(orders.router, prefix="/v2")  # versioning at include time

Graded Practice

Level 1 - Predict the Behavior

Given this endpoint:

@app.get("/search")
async def search(
    q: Annotated[str, Query(min_length=2, max_length=50)],
    page: Annotated[int, Query(ge=1, le=500)] = 1,
    size: Annotated[int, Query(ge=1, le=100)] = 20,
) -> dict:
    return {"q": q, "page": page, "size": size}

What HTTP status and body does each request return?

1. GET /search?q=py
2. GET /search?q=p
3. GET /search?q=python&page=0
4. GET /search?page=1&size=20
5. GET /search?q=python&page=1&size=200

Show Answer

1. 200 OK - {"q": "py", "page": 1, "size": 20}. All constraints satisfied; page and size use defaults.

2. 422 Unprocessable Entity - q has min_length=2 and "p" has length 1. FastAPI returns a validation error pointing to query → q with message about string length.

3. 422 Unprocessable Entity - page=0 violates ge=1. The error locates query → page.

4. 422 Unprocessable Entity - q is required (no default value). The error states that q is missing.

5. 422 Unprocessable Entity - size=200 violates le=100. The error locates query → size.

Level 2 - Debug This Code

A developer writes this endpoint to return a user's profile:

from sqlalchemy.orm import Session
from myapp.models import User  # SQLAlchemy ORM model

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)) -> User:
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404)
    return user

The endpoint works in testing. In production it raises MissingGreenlet errors intermittently, and sometimes returns JSON with password_hash included.

Identify all three bugs and write the fixed version.

Show Answer

Bug 1: Returning an ORM object directly as response_model. SQLAlchemy ORM objects are not JSON-serialisable. FastAPI calls Pydantic to serialise the return value, and Pydantic accesses ORM attributes - which may trigger lazy loading outside a session. In async contexts this causes MissingGreenlet errors (SQLAlchemy's async guard).

Bug 2: No response_model declared. Without response_model, FastAPI serialises whatever the handler returns. If the ORM object happens to serialise (e.g., using sqlalchemy-pydantic integration), it will include ALL columns - including password_hash. There is no filtering.

Bug 3: Using a synchronous Session in an async def handler. db.query(...) is synchronous SQLAlchemy. Inside async def, this blocks the event loop. Use AsyncSession and await db.execute(select(User)...).

Fixed version:

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from myapp.models import User
from myapp.schemas import UserResponse  # Pydantic model, no password_hash

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
) -> UserResponse:
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return UserResponse.model_validate(user)

UserResponse has model_config = ConfigDict(from_attributes=True) so model_validate reads ORM attributes correctly. password_hash is not in UserResponse, so it is never serialised.

Level 3 - Design Challenge

You are building a multi-tenant SaaS API. Each request must:

1. Extract the tenant ID from a subdomain header (X-Tenant-ID)
2. Look up the tenant in the database and verify it is active
3. Verify the Bearer JWT and decode the user ID
4. Verify the user belongs to the tenant
5. Rate-limit by tenant (100 req/minute)
6. Log the request with tenant_id, user_id, method, path, and duration

Design the dependency graph and middleware stack. Show which concerns belong in middleware vs Depends(), and why. Include code sketches for each component.

Show Answer

Middleware vs Depends - the division:

• Middleware: concerns that apply to every request without exception and that need access to the raw request before routing - request ID generation, structured logging with timing, rate limiting (IP-level). Middleware cannot easily access the dependency injection graph.
• Depends(): concerns that are business-logic-aware, that may vary per endpoint, that need database access, or that produce typed values the handler uses - tenant lookup, JWT validation, user-tenant verification.

Middleware stack (order: last added = outermost):

app.add_middleware(GZipMiddleware)
app.add_middleware(CORSMiddleware, allow_origins=["*"])

@app.middleware("http")
async def request_id_and_timing(request: Request, call_next):
    request.state.request_id = str(uuid.uuid4())
    start = time.perf_counter()
    response = await call_next(request)
    duration_ms = (time.perf_counter() - start) * 1000
    # Structured log - tenant_id/user_id set by dependencies via request.state
    logger.info({
        "request_id": request.state.request_id,
        "tenant_id": getattr(request.state, "tenant_id", None),
        "user_id": getattr(request.state, "user_id", None),
        "method": request.method,
        "path": request.url.path,
        "status": response.status_code,
        "duration_ms": round(duration_ms, 1),
    })
    response.headers["X-Request-ID"] = request.state.request_id
    return response

Dependency graph:

# Layer 1: parse tenant header
async def get_tenant(
    x_tenant_id: str = Header(...),
    db: AsyncSession = Depends(get_db),
) -> Tenant:
    tenant = await db.get(Tenant, x_tenant_id)
    if not tenant or not tenant.is_active:
        raise HTTPException(403, "Tenant not found or inactive")
    request.state.tenant_id = tenant.id  # for middleware logging
    return tenant

# Layer 2: validate JWT (depends on nothing except the header)
async def get_current_user(
    authorization: str = Header(...),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = authorization.removeprefix("Bearer ")
    payload = verify_jwt(token)  # raises 401 on invalid
    user = await db.get(User, payload["sub"])
    if not user:
        raise HTTPException(401, "User not found")
    request.state.user_id = user.id
    return user

# Layer 3: verify membership (depends on both above)
async def verify_membership(
    tenant: Tenant = Depends(get_tenant),
    user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
) -> tuple[Tenant, User]:
    membership = await db.execute(
        select(TenantMembership)
        .where(TenantMembership.tenant_id == tenant.id)
        .where(TenantMembership.user_id == user.id)
    )
    if not membership.scalar_one_or_none():
        raise HTTPException(403, "User does not belong to this tenant")
    return tenant, user

# Layer 4: rate limiting (depends on tenant)
async def rate_limit(
    tenant: Tenant = Depends(get_tenant),
    redis: Redis = Depends(get_redis),
) -> None:
    key = f"rate:{tenant.id}:{int(time.time() // 60)}"
    count = await redis.incr(key)
    await redis.expire(key, 60)
    if count > 100:
        raise HTTPException(429, "Rate limit exceeded")

# Apply to entire router
router = APIRouter(dependencies=[Depends(verify_membership), Depends(rate_limit)])

The rate-limiting dependency is at router level - it applies to every protected endpoint automatically. The timing and request ID are in middleware because they need to wrap the entire request lifecycle, including dependency resolution time.

Key Takeaways

• FastAPI = Starlette + Pydantic + type hints. Starlette provides the ASGI toolkit, Pydantic provides validation and serialisation, and FastAPI wires them together using Python type annotations. Understanding all three layers is required for debugging production issues.
• ASGI enables concurrency through coroutines, not threads. A single Uvicorn worker can handle thousands of concurrent I/O-bound requests. But async def is not magic - any blocking call inside async def blocks the event loop for every concurrent request. Use await for all I/O.
• def vs async def in FastAPI: def handlers run in a thread pool (safe, adds overhead). async def handlers run in the event loop (zero thread overhead, but must only use awaitable I/O). Never use synchronous blocking calls in async def.
• Annotated + Query/Path/Body is the canonical way to declare constraints. It keeps the type annotation and validation metadata together, is fully understood by type checkers, and generates accurate OpenAPI documentation.
• Depends(get_db) with yield is the standard pattern for database session management. Code before yield runs before the handler; code after yield runs after - including on exceptions, making it the correct place for commit/rollback logic.
• response_model on every endpoint is not optional in production. It strips sensitive fields (passwords, internal IDs), validates your handler's output during development, and generates accurate client-facing documentation.
• Never return ORM objects directly. Convert to a Pydantic response model with model_validate(orm_obj). ORM objects accessed outside a session cause DetachedInstanceError (sync) or MissingGreenlet (async).
• APIRouter with prefix and router-level dependencies is the correct way to organise a production API. Apply authentication and rate-limiting dependencies at the router level so they cannot be accidentally omitted from individual endpoints.
• TestClient for unit tests, httpx.AsyncClient for async integration tests. Use app.dependency_overrides to replace database sessions, external clients, and auth dependencies with test doubles - never hit a real database in unit tests.
• Disable /docs and /redoc in production unless your API is intentionally public. The OpenAPI schema reveals your internal data model, endpoint structure, and field names - useful for attackers.