FastAPI / Chapter 2 of 8

Validation, dependencies, and responses

Validate FastAPI requests with Pydantic, inject dependencies with Depends, and return response models. Includes APIRouter and async handlers.

7 topicsUpdated 22 Oct 2026

03 / 28

Path, query, body, and request metadata#

A name appearing in the URL template is a path parameter. Pydantic models are normally JSON bodies. Other simple parameters are normally query values. Use explicit markers for metadata or extra constraints.

PythonConstrained inputs

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

app = FastAPI()

class BookIn(BaseModel):
    title: str = Field(min_length=2, max_length=120)
    price: float = Field(gt=0)
    tags: list[str] = Field(default_factory=list)

@app.put("/books/{book_id}")
def put_book(
    book_id: Annotated[int, Path(ge=1)],
    book: BookIn,                          # JSON object body
    notify: Annotated[bool, Query()] = False,
    user_agent: Annotated[str | None, Header()] = None,
    session_id: Annotated[str | None, Cookie()] = None,
):
    return {"id": book_id, **book.model_dump(),
            "notify": notify, "agent": user_agent,
            "session_id": session_id}

@app.post("/search")
def search(filters: Annotated[dict, Body(embed=True)]):
    return filters  # expects {"filters": {...}}

HTTPMatching HTTP request

PUT /books/7?notify=true HTTP/1.1
Content-Type: application/json
User-Agent: demo-client
Cookie: session_id=abc

{"title":"A Fire Upon the Deep","price":19.99,"tags":["sf"]}

Common distinction

Header() maps user_agent to the User-Agent header. Query() can enforce min_length, ge, regex pattern, and more. A list query parameter accepts repeated keys such as ?tag=a&tag=b.

04 / 28

Pydantic: data models and validation#

A Pydantic model describes external data. Incoming values are parsed into Python values, invalid input produces 422, and output models stop private fields leaking into responses.

PythonSeparate input and output schemas

from datetime import datetime
from typing import Annotated, Literal
from pydantic import BaseModel, ConfigDict, EmailStr, Field

class UserCreate(BaseModel):
    email: EmailStr
    password: Annotated[str, Field(min_length=12)]

class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr
    joined_at: datetime

class Page(BaseModel):
    items: list[UserPublic]
    next_cursor: str | None = None

user = UserCreate.model_validate({"email": "a@b.com",
                                  "password": "long-password"})
print(user.model_dump(exclude={"password"}))
print(user.model_dump_json())

PythonValidation, safe updates, aliases

from pydantic import BaseModel, ConfigDict, Field, field_validator

class Product(BaseModel):
    model_config = ConfigDict(extra="forbid")
    sku: str = Field(pattern=r"^[A-Z]{2}-\d+$")
    price_cents: int = Field(ge=0)
    description: str | None = None

    @field_validator("sku", mode="before")
    @classmethod
    def normalize_sku(cls, value: str) -> str:
        return value.strip().upper()

class ProductPatch(BaseModel):
    price_cents: int | None = Field(default=None, ge=0)
    description: str | None = None

patch = ProductPatch.model_validate({"description": None})
changes = patch.model_dump(exclude_unset=True)
# changes == {"description": None}; omitted != explicit null

class Event(BaseModel):
    event_id: int = Field(alias="id")

Event.model_validate({"id": 3}).model_dump(by_alias=True)

Pydantic v2 names

Use model_validate, model_dump, model_dump_json, model_copy, field_validator, and ConfigDict(from_attributes=True). An optional type such as str | None remains required unless it has a default. EmailStr needs email-validator (included in some extras; install it explicitly if missing).

05 / 28

Advanced Pydantic: cross-field rules, unions, settings#

Use a model validator for rules involving several fields, a tagged union when one of several shapes is allowed, and pydantic-settings for environment configuration.

PythonCross-field check and tagged union

from typing import Annotated, Literal
from pydantic import BaseModel, Field, model_validator

class Card(BaseModel):
    kind: Literal["card"]
    last_four: str = Field(pattern=r"^\d{4}$")

class Bank(BaseModel):
    kind: Literal["bank"]
    account: str

Payment = Annotated[Card | Bank, Field(discriminator="kind")]

class Order(BaseModel):
    quantity: int = Field(gt=0)
    unit_price: int = Field(ge=0)
    payment: Payment
    discount: int = Field(default=0, ge=0)

    @model_validator(mode="after")
    def check_discount(self):
        if self.discount > self.quantity * self.unit_price:
            raise ValueError("Discount exceeds total")
        return self

PythonEnvironment-backed settings

# pip install pydantic-settings
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")
    database_url: str
    jwt_secret: str
    debug: bool = False

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

# DATABASE_URL and JWT_SECRET are required environment variables.
# Never place production secrets in committed .env files.

Output serialization

response_model asks FastAPI to validate/filter returned data. model_dump(exclude_unset=True) makes PATCH safe for omitted fields. Avoid accepting your ORM model directly as input: clients could set server-owned values.

06 / 28

Response models, errors, headers, and cookies#

State the response shape in the decorator. FastAPI validates and filters the returned value, generates a schema, and uses your declared status code.

PythonReturn a filtered response and Location header

from fastapi import FastAPI, HTTPException, Response, status
from pydantic import BaseModel

app = FastAPI()

class AccountOut(BaseModel):
    id: int
    email: str

@app.post("/accounts", response_model=AccountOut,
          status_code=status.HTTP_201_CREATED)
def create_account(response: Response):
    saved = {"id": 5, "email": "a@example.com", "password_hash": "secret"}
    response.headers["Location"] = f"/accounts/{saved['id']}"
    return saved  # password_hash is filtered out

@app.get("/accounts/{account_id}", response_model=AccountOut)
def get_account(account_id: int):
    if account_id != 5:
        raise HTTPException(status_code=404,
                            detail="Account not found")
    return {"id": 5, "email": "a@example.com"}

PythonExceptions and explicit JSON response

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

class DomainError(Exception):
    pass

@app.exception_handler(DomainError)
async def domain_error(request: Request, exc: DomainError):
    return JSONResponse(status_code=409,
                        content={"detail": "Conflict"})

# For explicit JSONResponse, provide JSON-safe data yourself;
# a response_model will not serialize/filter a returned Response.

Security boundary

Never rely on returned dicts to hide secrets by convention. A specific response_model prevents accidental extra fields. A response validation failure indicates a server bug; fix the returned data rather than masking it as a client error.

07 / 28

Dependencies: shared logic without repetition#

A dependency is a callable FastAPI runs before your route. Its return value becomes a parameter. Dependencies can themselves depend on others, and can clean up resources after a single yield.

PythonReusable dependency and aliases

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException, Query

app = FastAPI()

def require_api_key(x_api_key: Annotated[str | None, Header()] = None):
    if x_api_key != "example-key":
        raise HTTPException(401, "Missing or invalid API key")
    return x_api_key

ApiKey = Annotated[str, Depends(require_api_key)]

class Pagination:
    def __init__(self, max_size: int = 100):
        self.max_size = max_size

    def __call__(self, limit: Annotated[int, Query(ge=1)] = 20):
        return min(limit, self.max_size)

@app.get("/records")
def records(key: ApiKey, limit: Annotated[int, Depends(Pagination(50))]):
    return {"limit": limit}

PythonOne resource per request with yield

from collections.abc import Iterator
from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session, sessionmaker

engine = create_engine("sqlite:///./app.db")
SessionLocal = sessionmaker(engine)

def get_session() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session

Db = Annotated[Session, Depends(get_session)]
app = FastAPI()

@app.get("/health/db")
def check_db(db: Db):
    db.execute(text("SELECT 1"))
    return {"db_open": True}

Dependency behavior

FastAPI normally caches a dependency once per request. Use Depends(..., use_cache=False) when it must run again. yield cleanup runs on both success and error; let caught exceptions propagate. A streaming response may need its own resource lifecycle; consult the current scope rules before closing a session early.

08 / 28

APIRouter and application layout#

Keep endpoints by feature, then include each router in your app. Prefixes, tags, dependencies, and response metadata can be attached at the router level.

Pythonapp/books.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel

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

class Book(BaseModel):
    id: int
    title: str

@router.get("/{book_id}", response_model=Book)
def read_book(book_id: int):
    if book_id != 1:
        raise HTTPException(404, "Book not found")
    return Book(id=1, title="Dune")

Pythonapp/main.py

from fastapi import FastAPI
from app.books import router as books_router

app = FastAPI(title="Library API", version="1.0.0")
app.include_router(books_router, prefix="/v1")

@app.get("/health", include_in_schema=False)
def health():
    return {"ok": True}

# fastapi dev app/main.py
# GET /v1/books/1 → {"id":1,"title":"Dune"}

Module layout

A practical project can use app/main.py, app/books.py, app/models.py, app/schemas.py, app/db.py, and tests/. Keep imports acyclic; create the app in one place.

09 / 28

Async versus sync: which function to write#

async def can pause at await while other requests run. A normal def route is run in a thread pool. Async speeds up concurrent I/O when your entire call chain uses async libraries; it does not make CPU work faster.

PythonAsync I/O and synchronous I/O

import httpx
import requests
from fastapi import FastAPI

app = FastAPI()

@app.get("/async-weather")
async def async_weather():
    async with httpx.AsyncClient(timeout=5) as client:
        response = await client.get("https://example.org/weather")
        response.raise_for_status()
    return response.json()

@app.get("/sync-weather")
def sync_weather():
    response = requests.get("https://example.org/weather", timeout=5)
    response.raise_for_status()
    return response.json()

# Avoid requests.get() in async def: it blocks the event loop.

ShellThe scheduling idea

request A: await external_service() → yields to request B
request B: await database.fetch() → yields to request C
request C: does work → returns
request A: resumes when external_service completes

# CPU-heavy processing needs a separate process/worker queue;
# await alone cannot make it concurrent.

Call inside handler	Choose	Reason
`httpx.AsyncClient`, async DB driver	`async def` + `await`	Nonblocking I/O
Sync SQLAlchemy or blocking SDK	`def`	FastAPI moves route to thread pool
Heavy CPU or long job	Worker/process	Avoid holding a web worker

Database pairing

A sync Session belongs in sync handlers; an AsyncSession belongs in async handlers. Do not share either session across concurrent requests or tasks.