Learn Python Series):Last episode we got FastAPI up and running — routes, type validation, Pydantic models, auto-docs. That was the "Hello World" tier. Now let's build something real.
Real APIs need: passwords that meet complexity rules, email addresses that are actually valid, request data that passes business logic checks (not just type checks), shared authentication across dozens of endpoints, database sessions that open and close cleanly, and error responses that follow a consistent format. FastAPI handles all of this through two features: Pydantic validators and dependency injection.
Nota bene: "Dependency injection" sounds enterprise-Java-scary, but in FastAPI it's remarkably elegant. Think of it as "functions that run automatically before your endpoint, providing whatever it needs." Database connection? Dependency. Current authenticated user? Dependency. Rate limiter? Dependency. Pagination parameters? Dependency. Once this mental model clicks, you'll structure every serious API around it ;-)
In episode #49, we used Pydantic's built-in constraints (Field(gt=0, max_length=100)). But real-world validation often needs custom logic — password complexity rules, cross-field validation, data normalization. Pydantic v2 provides field_validator and model_validator for this:
from pydantic import BaseModel, Field, field_validator, model_validator
import re
class UserRegistration(BaseModel):
username: str = Field(..., min_length=3, max_length=30)
email: str
password: str = Field(..., min_length=8)
password_confirm: str
age: int = Field(..., ge=13) # Minimum age 13
@field_validator('username')
@classmethod
def username_must_be_alphanumeric(cls, v):
if not re.match(r'^[a-zA-Z][a-zA-Z0-9_]*$', v):
raise ValueError(
'Username must start with a letter and contain '
'only letters, numbers, and underscores'
)
return v.lower() # Normalize to lowercase
@field_validator('email')
@classmethod
def email_must_be_valid(cls, v):
# Basic email validation (use email-validator package for production)
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
if not re.match(pattern, v):
raise ValueError('Invalid email address format')
return v.lower()
@field_validator('password')
@classmethod
def password_complexity(cls, v):
if not re.search(r'[A-Z]', v):
raise ValueError('Password must contain at least one uppercase letter')
if not re.search(r'[a-z]', v):
raise ValueError('Password must contain at least one lowercase letter')
if not re.search(r'[0-9]', v):
raise ValueError('Password must contain at least one digit')
return v
@model_validator(mode='after')
def passwords_must_match(self):
if self.password != self.password_confirm:
raise ValueError('Passwords do not match')
return self
Several things to notice. @field_validator runs on individual fields — it receives the raw value and returns the (possibly transformed) value. You can normalize data here (lowercasing usernames, stripping whitespace). The @classmethod decorator is required in Pydantic v2.
@model_validator(mode='after') runs after all individual field validators have passed. It receives the fully constructed model instance, so you can check relationships between fields — like ensuring password and password_confirm match, or that end_date is after start_date.
If you need to validate before individual fields are processed, use mode='before':
class FlexibleInput(BaseModel):
values: list[int]
@model_validator(mode='before')
@classmethod
def coerce_single_to_list(cls, data):
"""Accept a single int or a list of ints."""
if isinstance(data, dict) and isinstance(data.get('values'), int):
data['values'] = [data['values']]
return data
This before validator sees the raw input data (a dict) before Pydantic tries to parse it. Useful for accepting flexible input formats.
When multiple models need the same validation logic, extract it into reusable functions:
from pydantic import field_validator
from typing import Any
def validate_not_empty(v: str) -> str:
"""Strip whitespace and reject empty strings."""
v = v.strip()
if not v:
raise ValueError('Field cannot be empty or whitespace-only')
return v
def validate_positive(v: int | float) -> int | float:
"""Reject non-positive numbers."""
if v <= 0:
raise ValueError('Must be a positive number')
return v
class Product(BaseModel):
name: str
description: str
price: float
quantity: int
# Apply reusable validators to specific fields
_validate_name = field_validator('name')(validate_not_empty)
_validate_description = field_validator('description')(validate_not_empty)
_validate_price = field_validator('price')(validate_positive)
_validate_quantity = field_validator('quantity')(validate_positive)
class Order(BaseModel):
customer_name: str
total: float
_validate_name = field_validator('customer_name')(validate_not_empty)
_validate_total = field_validator('total')(validate_positive)
Same validation logic, used across multiple models, zero code duplication. The functional approach (field_validator('field')(func)) might look unusual, but it's just calling the decorator manually instead of using @ syntax.
Sometimes you want fields that are derived from other fields — not stored, but calculated:
from pydantic import BaseModel, computed_field
class CartItem(BaseModel):
name: str
unit_price: float
quantity: int
tax_rate: float = 0.21 # 21% VAT (Dutch standard)
@computed_field
@property
def subtotal(self) -> float:
return round(self.unit_price * self.quantity, 2)
@computed_field
@property
def tax_amount(self) -> float:
return round(self.subtotal * self.tax_rate, 2)
@computed_field
@property
def total(self) -> float:
return round(self.subtotal + self.tax_amount, 2)
item = CartItem(name="Widget", unit_price=9.99, quantity=3)
print(item.model_dump())
# {'name': 'Widget', 'unit_price': 9.99, 'quantity': 3,
# 'tax_rate': 0.21, 'subtotal': 29.97, 'tax_amount': 6.29, 'total': 36.26}
Computed fields appear in serialization output (JSON responses, model_dump()) but are never accepted as input. They're pure derived values. This is much cleaner than calculating them in your endpoint code.
FastAPI provides specialized validators for each parameter source:
from fastapi import Query, Path, Header, Cookie
@app.get("/search/")
def search(
q: str = Query(
...,
min_length=2,
max_length=100,
description="Search query",
examples=["python fastapi", "web framework"],
),
page: int = Query(1, ge=1, le=1000, alias="p"),
per_page: int = Query(20, ge=1, le=100),
sort_by: str = Query("relevance", pattern="^(relevance|date|rating)$"),
):
return {
"query": q,
"page": page,
"per_page": per_page,
"sort_by": sort_by,
}
The alias="p" is subtle but powerful — clients send ?p=5 but your code reads page. The pattern parameter applies a regex constraint, so sort_by only accepts "relevance", "date", or "rating". And the examples show up in the OpenAPI docs.
For headers and cookies:
@app.get("/trace/")
def trace_request(
user_agent: str = Header(..., alias="User-Agent"),
session_id: str | None = Cookie(None),
x_request_id: str = Header(...),
):
return {
"user_agent": user_agent,
"session_id": session_id,
"request_id": x_request_id,
}
Note: HTTP headers use hyphens (X-Request-Id), but Python variables use underscores (x_request_id). FastAPI converts automatically. The alias override is needed for non-standard casing like User-Agent.
Dependencies in FastAPI are just functions (or callables) that your endpoint declares it needs. FastAPI calls them automatically before the endpoint runs:
from fastapi import Depends
def get_pagination(
skip: int = Query(0, ge=0, description="Records to skip"),
limit: int = Query(20, ge=1, le=100, description="Records per page"),
):
return {"skip": skip, "limit": limit}
@app.get("/items/")
def list_items(pagination: dict = Depends(get_pagination)):
# pagination = {"skip": 0, "limit": 20} (from query params)
all_items = list(range(100))
return {
"items": all_items[pagination["skip"]:pagination["skip"] + pagination["limit"]],
"pagination": pagination,
}
@app.get("/users/")
def list_users(pagination: dict = Depends(get_pagination)):
# Same pagination logic, reused
return {"users": [], "pagination": pagination}
FastAPI sees Depends(get_pagination), calls get_pagination with the request's query parameters, and passes the result to list_items as pagination. The dependency's parameters (skip, limit) are automatically extracted from the request — they appear in the OpenAPI docs, get validated, and have defaults. You get reusable, self-documenting request handling.
For dependencies that need configuration or state, use callable classes:
class Paginator:
def __init__(self, max_limit: int = 100):
self.max_limit = max_limit
def __call__(
self,
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1),
):
# Enforce max_limit regardless of what client requests
effective_limit = min(limit, self.max_limit)
return {"skip": skip, "limit": effective_limit}
# Different paginators for different endpoints
standard_paginator = Paginator(max_limit=100)
limited_paginator = Paginator(max_limit=10)
@app.get("/items/")
def list_items(pagination: dict = Depends(standard_paginator)):
return {"pagination": pagination}
@app.get("/expensive-resource/")
def list_expensive(pagination: dict = Depends(limited_paginator)):
return {"pagination": pagination} # Max 10 per page
The __call__ method makes the class instance callable. FastAPI treats it like any other dependency function. The instance holds configuration (max_limit), while each request gets its own __call__ invocation.
This is where dependency injection becomes truly powerful. Dependencies can depend on other dependencies, and FastAPI resolves the entire graph:
from fastapi import Header, HTTPException, Depends
# Layer 1: Extract token from header
def get_token(authorization: str = Header(...)):
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Invalid auth scheme")
return authorization[7:] # Strip "Bearer "
# Layer 2: Decode token into user data
def get_current_user(token: str = Depends(get_token)):
# In reality: decode JWT, query database
users_db = {
"valid-token-abc": {"id": 1, "username": "scipio", "role": "admin"},
"valid-token-xyz": {"id": 2, "username": "reader", "role": "viewer"},
}
user = users_db.get(token)
if not user:
raise HTTPException(status_code=401, detail="Invalid or expired token")
return user
# Layer 3: Require specific role
def require_role(required_role: str):
def dependency(user: dict = Depends(get_current_user)):
if user["role"] != required_role:
raise HTTPException(
status_code=403,
detail=f"Requires {required_role} role, you have {user['role']}"
)
return user
return dependency
# Usage: endpoints declare exactly what access level they need
@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
return {"user": user}
@app.delete("/users/{user_id}")
def delete_user(user_id: int, admin: dict = Depends(require_role("admin"))):
return {"deleted": user_id, "by": admin["username"]}
@app.get("/dashboard/")
def admin_dashboard(admin: dict = Depends(require_role("admin"))):
return {"stats": {"users": 42, "items": 100}}
The dependency chain flows: require_role("admin") → get_current_user → get_token → extracts from Authorization header. FastAPI resolves this automatically. If any layer fails (invalid token, wrong role), the appropriate HTTP error is returned and the endpoint never runs.
Notice require_role is a factory function — it returns a dependency function parameterized by the required role. This pattern (a function that returns a dependency) is how you create configurable dependencies.
yield: setup and teardownFor resources that need cleanup (database connections, file handles, external service clients), use yield in your dependency:
def get_db():
"""Provide a database session, ensure cleanup."""
db = SessionLocal() # Open connection
try:
yield db # Provide to endpoint
finally:
db.close() # Always close, even on error
def get_http_client():
"""Provide an HTTP client for external API calls."""
import httpx
client = httpx.Client(timeout=10.0)
try:
yield client
finally:
client.close()
@app.get("/users/")
def list_users(db = Depends(get_db)):
return db.query("SELECT * FROM users")
@app.get("/external-data/")
def get_external(client = Depends(get_http_client)):
response = client.get("https://api.example.com/data")
return response.json()
The code before yield is the setup (open connection). The code after yield (in finally) is the teardown (close connection). This runs even if the endpoint raises an exception — exactly like context managers (which we covered in episode #44).
In fact, this IS a context manager pattern integrated into FastAPI's dependency system. The beauty is that your endpoint function doesn't need to know about setup/teardown at all — it just receives db and uses it. Separation of concerns.
Apply dependencies to all endpoints at once:
# Apply to entire application
app = FastAPI(dependencies=[Depends(verify_api_key)])
# Apply to a specific router
from fastapi import APIRouter
admin_router = APIRouter(
prefix="/admin",
tags=["admin"],
dependencies=[Depends(require_role("admin"))],
)
@admin_router.get("/stats")
def admin_stats():
# require_role("admin") already ran — no need to declare it again
return {"total_users": 42}
@admin_router.post("/broadcast")
def send_broadcast(message: str):
return {"sent": True, "message": message}
app.include_router(admin_router)
Every endpoint under /admin/ automatically requires admin authentication. The individual endpoint functions don't even mention authentication — it's handled at the router level.
FastAPI handles file uploads through UploadFile, which is memory-efficient (large files are spooled to disk, not loaded entirely into memory):
from fastapi import File, UploadFile, HTTPException
ALLOWED_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"}
MAX_SIZE = 5 * 1024 * 1024 # 5 MB
@app.post("/upload/avatar")
async def upload_avatar(
file: UploadFile = File(..., description="Profile picture (max 5MB)"),
user: dict = Depends(get_current_user),
):
# Validate content type
if file.content_type not in ALLOWED_TYPES:
raise HTTPException(
status_code=415,
detail=f"Unsupported file type: {file.content_type}. "
f"Allowed: {', '.join(ALLOWED_TYPES)}"
)
# Read file content
contents = await file.read()
# Validate file size
if len(contents) > MAX_SIZE:
raise HTTPException(
status_code=413,
detail=f"File too large: {len(contents)} bytes. Max: {MAX_SIZE}"
)
# In reality: save to disk or cloud storage
return {
"filename": file.filename,
"content_type": file.content_type,
"size_bytes": len(contents),
"user": user["username"],
}
@app.post("/upload/documents")
async def upload_documents(files: list[UploadFile] = File(...)):
results = []
for f in files:
content = await f.read()
results.append({
"filename": f.filename,
"size": len(content),
})
return {"uploaded": len(results), "files": results}
UploadFile provides async methods (read(), write(), seek(), close()) because file I/O can be slow. The await file.read() doesn't block the event loop while reading large uploads.
For HTML form submissions (content type application/x-www-form-urlencoded instead of application/json):
from fastapi import Form
@app.post("/login/")
def login(
username: str = Form(..., min_length=3),
password: str = Form(..., min_length=8),
remember_me: bool = Form(False),
):
# In reality: verify credentials against database
return {
"username": username,
"remember_me": remember_me,
"message": "Login successful",
}
Important: you cannot mix Form and JSON Body in the same endpoint. A request is either form-encoded or JSON-encoded — not both. This is an HTTP protocol constraint, not a FastAPI limitation.
You can combine Form and File in the same endpoint though (for multipart form submissions):
@app.post("/submit-report/")
async def submit_report(
title: str = Form(...),
description: str = Form(...),
attachment: UploadFile = File(None), # Optional file
):
result = {"title": title, "description": description}
if attachment:
content = await attachment.read()
result["attachment_size"] = len(content)
return result
By default, FastAPI returns validation errors in Pydantic's format. You can customize this globally:
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(
request: Request, exc: RequestValidationError
):
"""Custom format for validation errors."""
errors = []
for error in exc.errors():
errors.append({
"field": " → ".join(str(loc) for loc in error["loc"]),
"message": error["msg"],
"type": error["type"],
})
return JSONResponse(
status_code=422,
content={
"error": "validation_error",
"detail": f"{len(errors)} validation error(s)",
"errors": errors,
},
)
Now every validation error across your entire API uses this consistent format. Your API consumers get predictable error responses they can parse programmatically.
One of dependency injection's biggest benefits: in tests, you can swap real dependencies for fakes:
from fastapi.testclient import TestClient
# Real dependency: queries actual database
def get_db():
db = RealDatabase()
try:
yield db
finally:
db.close()
# Test override: uses in-memory fake
def get_fake_db():
return FakeDatabase({"users": [{"id": 1, "name": "Test User"}]})
# Override for testing
app.dependency_overrides[get_db] = get_fake_db
client = TestClient(app)
def test_list_users():
response = client.get("/users/")
assert response.status_code == 200
assert len(response.json()["users"]) == 1
# Override auth for testing protected endpoints
def fake_current_user():
return {"id": 1, "username": "test", "role": "admin"}
app.dependency_overrides[get_current_user] = fake_current_user
def test_admin_endpoint():
response = client.get("/admin/stats")
assert response.status_code == 200
# Clean up overrides after testing
app.dependency_overrides.clear()
No mocking libraries needed. No monkey-patching. Just provide a different function with the same return type. This is one of the clearest testing patterns I've ever used — and it works because dependencies are functions with explicit contracts, not hidden global state.
Let me show how all these pieces compose into a realistic (if small) application:
from fastapi import FastAPI, Depends, HTTPException, Query
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
app = FastAPI(title="Bookshelf API")
# --- Models ---
class BookCreate(BaseModel):
title: str = Field(..., min_length=1, max_length=200)
author: str = Field(..., min_length=1, max_length=100)
isbn: str = Field(..., pattern=r'^\d{13}$')
year: int = Field(..., ge=1450, le=datetime.now().year)
@field_validator('title', 'author')
@classmethod
def strip_whitespace(cls, v):
return v.strip()
class BookResponse(BookCreate):
id: int
created_at: datetime
# --- Database (in-memory for demo) ---
books_db: dict[int, dict] = {}
next_id = 1
# --- Dependencies ---
def get_book_or_404(book_id: int):
if book_id not in books_db:
raise HTTPException(status_code=404, detail=f"Book {book_id} not found")
return books_db[book_id]
# --- Endpoints ---
@app.post("/books/", response_model=BookResponse, status_code=201)
def create_book(book: BookCreate):
global next_id
db_book = {**book.model_dump(), "id": next_id, "created_at": datetime.now()}
books_db[next_id] = db_book
next_id += 1
return db_book
@app.get("/books/", response_model=list[BookResponse])
def list_books(
skip: int = Query(0, ge=0),
limit: int = Query(20, ge=1, le=100),
author: str | None = Query(None, min_length=2),
):
results = list(books_db.values())
if author:
results = [b for b in results if author.lower() in b["author"].lower()]
return results[skip:skip + limit]
@app.get("/books/{book_id}", response_model=BookResponse)
def get_book(book: dict = Depends(get_book_or_404)):
return book
@app.delete("/books/{book_id}", status_code=204)
def delete_book(book_id: int, book: dict = Depends(get_book_or_404)):
del books_db[book_id]
Notice: get_book_or_404 is a dependency that both validates the book exists AND returns it. The get_book and delete_book endpoints both use it, so the "not found" logic is written once. The BookCreate model handles all input validation. The BookResponse model controls what gets returned. And all of this generates complete OpenAPI documentation at /docs.
In this episode, we explored advanced FastAPI validation and dependency injection:
@field_validator enables custom validation logic per field with data normalization@model_validator validates relationships between fields (password confirmation, date ranges)@computed_field adds derived read-only fields to Pydantic modelsQuery, Path, Header, Cookie provide source-specific validation with aliases, patterns, and constraintsrequire_role("admin")) create parameterized dependenciesyield dependencies handle setup/teardown like context managersUploadFile handles file uploads with async, memory-efficient streamingdependency_overrides enables clean testing without mockingDependencies are FastAPI's secret weapon. Once you start thinking in dependencies, you stop writing monolithic endpoint functions with 200 lines of setup logic. Your endpoints become what they should be — thin handlers that declare what they need and focus on business logic. Everything else is composed from reusable parts ;-)
@scipio