Pydantic v2 Deep Dive
IntermediatePydantic v2 (used by FastAPI 0.100+) was completely rewritten in Rust, making it 5–50× faster than v1. The API changed significantly: validators use new decorators, configuration uses a new class, and several methods were renamed. Understanding these changes is essential for reading modern FastAPI code and migrating older projects.
from pydantic import (
BaseModel, Field, field_validator, model_validator,
computed_field, ConfigDict
)
from decimal import Decimal
class Product(BaseModel):
# ConfigDict replaces the old "class Config" inner class
model_config = ConfigDict(
str_strip_whitespace=True, # auto-strip spaces: " hello " → "hello"
validate_assignment=True, # re-validate when you do product.price = -1
populate_by_name=True, # allow both alias AND field name in JSON
)
# Field() adds constraints + documentation
name: str = Field(..., min_length=2, max_length=100) # ... = required
price: Decimal = Field(gt=0, decimal_places=2) # must be positive
tags: list[str] = Field(default_factory=list) # mutable default (IMPORTANT)
# @field_validator: validate/transform a single field
# @classmethod is REQUIRED in v2 (was optional in v1)
@field_validator("name")
@classmethod
def name_no_digits(cls, v: str) -> str:
if any(c.isdigit() for c in v):
raise ValueError("Product name must not contain digits")
return v.title() # transform: "running shoes" → "Running Shoes"
# @model_validator: cross-field logic (runs after all fields are set)
@model_validator(mode="after")
def check_sale_price(self) -> "Product":
if "sale" in self.tags and self.price > 100:
raise ValueError("Sale items must be under $100")
return self # must return self in mode="after"
# @computed_field: a derived property included in serialization
@computed_field
@property
def price_with_tax(self) -> Decimal:
return self.price * Decimal("1.1")
# --- Serialization ---
p = Product(name=" running shoes ", price="49.99", tags=["sale"])
# name was stripped and title-cased: "Running Shoes"
p.model_dump() # → dict (v1: .dict())
p.model_dump_json() # → JSON string (v1: .json())
p.model_dump(by_alias=True) # use alias keys instead of field names
p.model_dump(exclude_unset=True) # only fields explicitly set by caller
p.model_dump(exclude={"price"}) # exclude specific fields
# Partial updates: only fields the client sent
class ProductUpdate(BaseModel):
name: str | None = None
price: Decimal | None = None
update = ProductUpdate(price="59.99")
update.model_dump(exclude_unset=True) # {"price": 59.99} — name not includedReplaces the old class Config: inner class from Pydantic v1. ConfigDict is a typed dict so your IDE can autocomplete all available options. The outer assignment model_config = is read by Pydantic at class creation time.
In Pydantic v2, all field validators must be class methods (decorated with @classmethod). The cls parameter receives the model class. The function must return the (possibly transformed) value — if you raise ValueError, Pydantic converts it to a validation error.
model_validator(mode="after") receives the fully constructed model instance — all fields are already validated. mode="before" receives the raw input dict before any field validation. Use “after” for cross-field checks; use “before” to reshape incoming data.
A Python @property that is included in model_dump() and JSON output. Without @computed_field, properties are just Python properties and won’t appear in API responses. Great for derived values like full_name, price_with_tax.
Never use Field(default=[]) — a mutable default is shared across all instances (a classic Python bug). default_factory=list calls list() fresh for each new instance, giving each its own empty list.
Only includes fields that were explicitly set by the caller, not fields that have defaults. Critical for PATCH endpoints: if client sends {"price": 9.99}, you get {"price": 9.99} back — not {"name": None, "price": 9.99, ...}.
⚠ Key v1 → v2 Migration Changes
@validator→@field_validator(must add@classmethod)@root_validator→@model_validator(mode="before"|"after").dict()→.model_dump()(old still works, but deprecated).json()→.model_dump_json()class Config:→model_config = ConfigDict(...)schema()→model_json_schema()
Dependency Injection System
IntermediateDependency Injection (DI) is FastAPI’s most powerful feature. You declare shared logic — database sessions, authentication, pagination — as functions, then inject them into route handlers with Depends(). FastAPI automatically calls each dependency, passes the result to your handler, and even supports cleanup code that runs after the response is sent (using yield).
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Annotated
app = FastAPI()
# ── Simple function dependency ───────────────────────────────────────────────
# FastAPI sees Depends(get_pagination) and calls get_pagination()
# injecting its return value as the 'pagination' parameter
def get_pagination(skip: int = 0, limit: int = 20):
return {"skip": skip, "limit": min(limit, 100)} # cap at 100
@app.get("/items")
def list_items(pagination: Annotated[dict, Depends(get_pagination)]):
# pagination = {"skip": 0, "limit": 20} ← injected automatically
return pagination
# ── Class-based dependency (when you need state or methods) ─────────────────
class QueryFilter:
def __init__(self, q: str | None = None, active: bool = True):
self.q = q
self.active = active
# Depends() with NO argument infers the type annotation and calls it
@app.get("/users")
def list_users(filters: Annotated[QueryFilter, Depends()]):
return {"q": filters.q, "active": filters.active}
# ── yield dependency — cleanup after response is sent ────────────────────────
# This is the canonical pattern for database sessions
def get_db():
db = {"session": "open"} # in real code: SessionLocal()
try:
yield db # ← route handler receives db here
finally:
pass # db.close() ← runs AFTER response is sent to client
# ── Chaining: dependencies that depend on other dependencies ─────────────────
def get_current_user(
token: str | None = Header(default=None, alias="x-token"),
db: Annotated[dict, Depends(get_db)] = None,
):
if not token or token != "valid-token":
raise HTTPException(status_code=401, detail="Invalid token")
return {"user_id": 1, "username": "alice"}
@app.get("/profile")
def get_profile(
user: Annotated[dict, Depends(get_current_user)] # chains auth + db
):
return user
# ── Route-level dependency (no return value needed — side effect only) ───────
def check_api_key(x_api_key: str = Header(...)):
if x_api_key != "secret-key":
raise HTTPException(status_code=403, detail="Invalid API key")
# No return needed — just raises or passes
@app.get("/data", dependencies=[Depends(check_api_key)]) # applied, not injected
def get_data():
return {"data": "protected"}Tells FastAPI: “call get_pagination before calling this route handler, and pass its return value here.” FastAPI inspects get_pagination’s own parameters (like skip and limit) and resolves those from the request too — recursively.
Using yield turns the dependency into a context manager. Code before yield runs when the request starts. The yielded value is injected into the route. Code after yield (in finally) runs after the response is fully sent — guaranteed even if the route raises an exception.
The modern Pydantic v2 style. Annotated[Type, Depends(fn)] attaches the dependency metadata to the type hint. When Depends() has no argument, FastAPI infers the callable from the type annotation (useful for class-based deps).
By default, if the same dependency is used multiple times in one request, FastAPI calls it once and reuses the result. This is why a DB session opened in get_db is the same session object everywhere it’s used in that request.
When you don’t need the dependency’s return value (just its side effects like authentication checks), use the dependencies= list on the decorator. The dependency runs, but its result is discarded. Also available on APIRouter() to protect entire router sections.
💬 Interview Tip
“How does FastAPI’s DI system work?” — “FastAPI inspects function signatures at startup using Python’s inspect module to build a dependency graph. At request time, it resolves the graph by calling each dependency in topological order. The same dependency called multiple times per request is cached (called once, value reused). Generators with yield provide resource lifecycle management — cleanup code runs after the response, guaranteed via finally.”
APIRouter & App Structure
IntermediateAPIRouter is a mini-version of the main FastAPI app. You define routes in separate modules (one per resource), then register them on the main app with include_router(). This keeps each router file small and focused, and lets you apply shared settings (prefix, tags, auth) to entire groups of routes at once.
# routers/items.py
from fastapi import APIRouter, Depends, HTTPException
from app.schemas import ItemCreate, ItemRead
from app.dependencies import get_current_user
# APIRouter acts like a mini FastAPI app
# prefix: prepended to all paths in this router (GET /items/, GET /items/{id})
# tags: groups all endpoints under "items" in Swagger docs
router = APIRouter(
prefix="/items",
tags=["items"],
responses={404: {"description": "Item not found"}}, # shared across all routes
)
@router.get("/") # actual URL: GET /items/
def list_items():
return [{"id": 1, "name": "Widget"}]
@router.get("/{item_id}") # actual URL: GET /items/{item_id}
def get_item(item_id: int):
if item_id > 100:
raise HTTPException(status_code=404, detail="Item not found")
return {"id": item_id}
@router.post("/", status_code=201)
def create_item(item: ItemCreate):
return {"id": 99, **item.model_dump()}# app/main.py
from fastapi import FastAPI, Depends
from app.routers import items, users, orders
from app.dependencies import verify_api_key
app = FastAPI(title="My API")
# include_router registers all routes from the router onto the main app
app.include_router(items.router)
app.include_router(users.router)
# You can override or extend router settings at include time:
app.include_router(
orders.router,
prefix="/api/v2", # changes prefix: /orders → /api/v2/orders
dependencies=[Depends(verify_api_key)], # adds auth to ALL order routes
)
# Routers can have sub-routers (nested routing)
# admin_router includes user_admin_router as /admin/users
from app.routers import admin
app.include_router(admin.router, prefix="/admin", tags=["admin"])The prefix is prepended to every route in this router. So @router.get("/") becomes GET /items/ and @router.get("/{id}") becomes GET /items/{id}. You never repeat the prefix on individual routes.
This is where the router’s routes are copied onto the main app’s route table. Until include_router() is called, the router exists in isolation. You can include the same router multiple times with different prefixes to version your API.
Setting dependencies=[Depends(auth)] on either APIRouter() or include_router() applies that dependency to every route in the router. This is the clean way to require authentication for an entire section of your API without adding Depends(auth) to every single route function.
Documents shared error responses that apply to all routes in the router. This appears in the OpenAPI spec and Swagger docs, telling clients “any route in this router may return a 404.” It’s documentation only — it doesn’t add any runtime behavior.
Middleware
IntermediateMiddleware is code that runs before every request reaches your route handler, and after every response leaves it. Think of it as a “wrapper” around all your routes. Common uses: adding headers, measuring timing, logging, injecting request IDs, authentication on all routes. FastAPI supports both function-based and class-based middleware.
import time, uuid
from fastapi import FastAPI, Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
app = FastAPI()
# ── Function-based middleware (simplest) ────────────────────────────────────
@app.middleware("http")
async def add_timing_header(request: Request, call_next):
# Code here runs BEFORE the route handler
start = time.perf_counter()
response: Response = await call_next(request) # ← calls the actual route
# Code here runs AFTER the route handler (but before sending to client)
elapsed = time.perf_counter() - start
response.headers["X-Process-Time"] = f"{elapsed:.4f}s"
return response
# ── Class-based middleware (better for complex logic / state) ────────────────
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request_id = str(uuid.uuid4())
# request.state is a namespace for request-scoped data
# accessible from anywhere in this request's lifecycle
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
app.add_middleware(RequestIDMiddleware)
# ── Built-in Starlette middleware ────────────────────────────────────────────
from starlette.middleware.gzip import GZipMiddleware
from starlette.middleware.trustedhost import TrustedHostMiddleware
# Compress responses larger than 1000 bytes
app.add_middleware(GZipMiddleware, minimum_size=1000)
# Reject requests with bad Host headers (security)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["example.com", "*.example.com"])
# Middleware execution order: LAST added = FIRST to run on request
# add_middleware(A) then add_middleware(B)
# Request flow: B → A → route handler → A → BThis is the critical line — it passes the request to the next layer (either the next middleware or the actual route handler). Everything before this runs on the way IN; everything after runs on the way OUT. If you forget to call it, the request never reaches your routes.
A simple namespace object attached to each request. You can store arbitrary data here (request ID, authenticated user, start time) and access it from any middleware, dependency, or route handler processing that same request.
Middleware is applied in reverse order of how you add it (LIFO — Last In, First Out). The last middleware added is the outermost wrapper, so it runs first on the request and last on the response. Keep this in mind when ordering matters (e.g., auth before logging).
You can modify or add response headers in middleware after await call_next(request). This is how you add standard headers like X-Request-ID, X-Process-Time, or CORS headers across all responses without touching individual routes.
⚠ Gotcha — BaseHTTPMiddleware breaks streaming
BaseHTTPMiddleware buffers the entire response body in memory before passing it along. This completely breaks streaming responses (topic 29) and SSE (topic 30). For apps that stream, use a pure ASGI middleware (a class implementing __call__(scope, receive, send)) or handle only request-side logic in your middleware.
CORS Configuration
IntermediateCORS (Cross-Origin Resource Sharing) is a browser security feature. When your React app at localhost:3000 tries to call your API at localhost:8000, the browser blocks the request unless the API explicitly says “I allow this origin.” FastAPI uses Starlette’s CORSMiddleware to handle this.
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
# ── Development: permissive (allow your local frontends) ─────────────────────
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "http://localhost:5173"], # Vite/CRA
allow_credentials=True, # required if sending cookies or Authorization headers
allow_methods=["*"], # allow all HTTP methods
allow_headers=["*"], # allow all headers
)
# ── Production: restrictive (only your actual domains) ───────────────────────
ALLOWED_ORIGINS = [
"https://app.example.com",
"https://www.example.com",
]
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS, # explicit list — no wildcards in production
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH", "OPTIONS"],
allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
max_age=3600, # how long browser can cache the preflight response (seconds)
)The list of frontend origins allowed to make cross-origin requests. An “origin” is scheme + host + port: http://localhost:3000 is different from https://localhost:3000. The middleware adds Access-Control-Allow-Origin headers based on this list.
Required when the browser sends cookies or Authorization headers with the request. This adds Access-Control-Allow-Credentials: true to responses. Without it, the browser strips credentials from cross-origin requests.
Before a non-simple request (e.g., POST with JSON), the browser automatically sends an OPTIONS preflight request to ask “do you allow this?” CORSMiddleware handles these automatically — you don’t need a separate @app.options() route.
Tells the browser it can cache the preflight response for 3600 seconds (1 hour). Without this, the browser sends a preflight before every single cross-origin request, adding latency. Setting a reasonable max_age is a performance optimization.
⚠ Gotcha — wildcard + credentials = browser error
Setting allow_origins=["*"] combined with allow_credentials=True is rejected by browsers — it’s a CORS spec restriction (not a FastAPI bug). When using credentials, you must specify explicit origins. Use the environment-based list pattern above.
Background Tasks
IntermediateBackground tasks let you run code after the HTTP response is already sent to the client. Ideal for “fire and forget” operations like sending welcome emails or logging — things that shouldn’t block the response. FastAPI injects a BackgroundTasks object; you add functions to it; FastAPI runs them after the response.
from fastapi import FastAPI, BackgroundTasks
import logging, time
app = FastAPI()
logger = logging.getLogger(__name__)
# ── Define a task function (sync or async, both work) ────────────────────────
def send_welcome_email(email: str, username: str):
# This runs AFTER the HTTP response has been sent to the client
# The client already got their response — they're not waiting for this
time.sleep(2) # simulate slow email sending
logger.info(f"Welcome email sent to {email}")
async def log_activity(user_id: int, action: str):
# async tasks also work — they run on the event loop
logger.info(f"User {user_id} performed: {action}")
# ── Inject BackgroundTasks and schedule work ─────────────────────────────────
@app.post("/users/register")
def register_user(
email: str,
username: str,
background_tasks: BackgroundTasks, # FastAPI injects this automatically
):
# 1. Do the important work (synchronous, before response)
user = {"id": 1, "email": email, "username": username}
# Save to DB here...
# 2. Schedule side effects to run after response is sent
background_tasks.add_task(send_welcome_email, email, username)
background_tasks.add_task(log_activity, user["id"], "registered")
# 3. Return response immediately — client doesn't wait for email/logging
return {"message": "Registration successful!", "user_id": user["id"]}
# ── Background tasks from dependencies ───────────────────────────────────────
from fastapi import Depends
def get_bg(background_tasks: BackgroundTasks):
return background_tasks
@app.get("/items/{item_id}")
def get_item(item_id: int, bg: BackgroundTasks = Depends()):
bg.add_task(log_activity, 0, f"viewed item {item_id}")
return {"item_id": item_id}FastAPI detects this parameter type and injects a BackgroundTasks instance automatically — no Depends() needed. It’s one of the few “magic” parameter types, alongside Request and Response.
Queues the function to run after the response is sent. Arguments are positional and keyword — add_task(fn, email, username) calls fn(email, username). You can add multiple tasks; they run sequentially in the order added.
Background tasks run in the same server process, after Starlette sends the HTTP response. The client receives their response immediately; the task runs a fraction of a second later. This improves perceived response time without true parallel execution.
⚠ Gotcha — Background tasks are NOT for heavy work
Background tasks run in the same process and block other requests if they’re CPU-intensive or slow. For heavy operations (PDF generation, sending 10,000 emails, ML inference), use a real job queue: Celery, ARQ, or FastAPI-TaskIQ (see Topic 51). Background tasks are for lightweight fire-and-forget work only.
WebSockets
IntermediateWebSockets provide a persistent, full-duplex connection — both client and server can send messages at any time. Unlike HTTP (one request → one response), WebSockets stay open. Use them for real-time features: chat, live notifications, collaborative editing, or game state. FastAPI supports WebSockets natively via Starlette.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
# ── Simple echo WebSocket ─────────────────────────────────────────────────────
@app.websocket("/ws") # note: @app.websocket, not @app.get
async def websocket_echo(websocket: WebSocket):
await websocket.accept() # MUST accept before sending/receiving — handshake
try:
while True:
data = await websocket.receive_text() # blocks until client sends
await websocket.send_text(f"Echo: {data}")
except WebSocketDisconnect:
pass # client closed connection — handle gracefully
# ── Connection manager for multiple clients / broadcasting ───────────────────
class ConnectionManager:
def __init__(self):
self.connections: list[WebSocket] = []
async def connect(self, ws: WebSocket):
await ws.accept()
self.connections.append(ws)
def disconnect(self, ws: WebSocket):
self.connections.remove(ws)
async def broadcast(self, message: str):
# Send to all connected clients
dead = []
for ws in self.connections:
try:
await ws.send_text(message)
except Exception:
dead.append(ws) # remove broken connections
for ws in dead:
self.connections.remove(ws)
manager = ConnectionManager()
@app.websocket("/ws/chat/{room}")
async def chat(
websocket: WebSocket,
room: str,
token: str | None = None, # query params work: ws://host/ws/chat/room?token=...
):
await manager.connect(websocket)
try:
while True:
message = await websocket.receive_text()
await manager.broadcast(f"[{room}] {message}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast(f"[{room}] A user left")Completes the WebSocket upgrade handshake. You must call this before sending or receiving any messages. Forgetting it causes cryptic errors. The connection starts as an HTTP upgrade request; accept() confirms the upgrade to WebSocket protocol.
WebSocket handlers run in an infinite loop — they keep the connection open and wait for messages. await websocket.receive_text() suspends the coroutine until the client sends something, freeing the event loop for other connections.
Raised when the client closes the connection or drops off. If you don’t catch this, FastAPI logs an unhandled exception. Always wrap your receive loop in try/except WebSocketDisconnect and do cleanup (remove from connection list, notify others).
A pattern to manage multiple connections. Since WebSocket state is in-memory, this only works on a single instance. For multiple server instances, you need Redis Pub/Sub to broadcast across instances (see Topic 52).
💬 Interview Tip
When asked “WebSockets vs SSE vs HTTP polling?”: WebSockets = bidirectional, persistent, complex setup; SSE = server→client only, simpler, HTTP/1.1 compatible, auto-reconnect; Long polling = client waits for response, legacy pattern. Use WebSockets for chat/games, SSE for live feeds/notifications.
Exception Handling & Custom Error Responses
IntermediateBy default, FastAPI returns generic error shapes. Real APIs need consistent, structured error responses. You can register global exception handlers for any exception type — including your own domain exceptions — using @app.exception_handler(). You can also override FastAPI’s default 422 validation error format.
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
app = FastAPI()
# ── Define domain-specific exceptions ────────────────────────────────────────
class ResourceNotFoundError(Exception):
"""Raised when a requested resource doesn't exist."""
def __init__(self, resource: str, resource_id: int):
self.resource = resource
self.resource_id = resource_id
class DuplicateResourceError(Exception):
"""Raised when trying to create something that already exists."""
def __init__(self, field: str, value: str):
self.field = field
self.value = value
# ── Register handlers for your custom exceptions ─────────────────────────────
@app.exception_handler(ResourceNotFoundError)
async def not_found_handler(request: Request, exc: ResourceNotFoundError):
return JSONResponse(
status_code=404,
content={
"error": "RESOURCE_NOT_FOUND",
"message": f"{exc.resource} with id={exc.resource_id} does not exist",
"path": str(request.url.path),
},
)
@app.exception_handler(DuplicateResourceError)
async def duplicate_handler(request: Request, exc: DuplicateResourceError):
return JSONResponse(
status_code=409,
content={
"error": "DUPLICATE_RESOURCE",
"message": f"A resource with {exc.field}='{exc.value}' already exists",
},
)
# ── Override FastAPI's default 422 validation error format ───────────────────
@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={
"error": "VALIDATION_ERROR",
"detail": exc.errors(), # list of field-level errors with loc, msg, type
},
)
# ── Use domain exceptions in routes ──────────────────────────────────────────
fake_db = {1: {"name": "Alice"}, 2: {"name": "Bob"}}
@app.get("/users/{user_id}")
def get_user(user_id: int):
if user_id not in fake_db:
raise ResourceNotFoundError("User", user_id) # raise domain exception directly
return fake_db[user_id]
@app.post("/users/")
def create_user(email: str):
if email in ["alice@x.com"]:
raise DuplicateResourceError("email", email)
return {"message": "Created"}Raising your own class ResourceNotFoundError(Exception) instead of HTTPException keeps your business logic layer free of HTTP concerns. The service or repository raises a domain error; the handler translates it to HTTP. Services should know nothing about status codes.
Registers a function to handle any uncaught exception of that type in your app. FastAPI catches the exception, calls your handler, and uses the returned JSONResponse as the HTTP response. This centralizes error formatting in one place.
FastAPI’s own exception for Pydantic validation failures (wrong types, missing fields, constraint violations). By overriding its handler, you can reshape the error format to match your API’s error schema instead of FastAPI’s default format.
Returns a list of dicts, each describing one validation failure: {"loc": ["body", "price"], "msg": "value is not a valid float", "type": "type_error.float"}. The loc field is a path to the invalid field, making field-level error highlighting easy in frontend UIs.
★ Rarely Known — exc.body
RequestValidationError has an exc.body attribute containing the raw request body that failed validation. Extremely useful for debugging (log exactly what the client sent) but be careful not to log sensitive fields (passwords, tokens) in production.
Authentication — HTTP Basic & API Keys
IntermediateBefore diving into JWT (Topic 20), understand the simpler authentication schemes. HTTP Basic sends username/password in the Authorization: Basic base64(user:pass) header. API Keys send a token in a header, query param, or cookie. Both are simpler than JWT but appropriate for server-to-server communication or simple admin APIs.
import secrets
from fastapi import FastAPI, Depends, HTTPException, Security, status
from fastapi.security import (
HTTPBasic, HTTPBasicCredentials,
APIKeyHeader, APIKeyQuery,
)
from typing import Annotated
app = FastAPI()
basic_auth = HTTPBasic()
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
api_key_query = APIKeyQuery(name="api_key", auto_error=False)
VALID_API_KEY = "my-production-secret"
# ── HTTP Basic Auth ──────────────────────────────────────────────────────────
@app.get("/basic/protected")
def basic_protected(
credentials: Annotated[HTTPBasicCredentials, Depends(basic_auth)]
):
# CRITICAL: use secrets.compare_digest, not == (prevents timing attacks)
is_valid_user = secrets.compare_digest(credentials.username, "admin")
is_valid_pass = secrets.compare_digest(credentials.password, "secret123")
if not (is_valid_user and is_valid_pass):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": "Basic"}, # prompts browser login dialog
)
return {"user": credentials.username, "authenticated": True}
# ── API Key Auth (from header OR query string) ───────────────────────────────
async def get_api_key(
header_key: str | None = Security(api_key_header),
query_key: str | None = Security(api_key_query),
) -> str:
key = header_key or query_key
if key and secrets.compare_digest(key, VALID_API_KEY):
return key
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid or missing API key",
)
@app.get("/api/data")
def get_data(api_key: str = Depends(get_api_key)):
return {"data": "secret", "authenticated_with": api_key[:4] + "..."}Python’s string == operator short-circuits — it returns as soon as it finds a mismatch. An attacker can measure how long comparisons take to guess credentials character by character (timing attack). compare_digest takes the same time regardless of where strings differ, preventing this.
Starlette automatically decodes the Authorization: Basic base64(user:pass) header and gives you a credentials object with .username and .password as plain strings. You never need to decode Base64 yourself.
By default, if the security scheme doesn’t find its credential, it raises a 403 automatically. Setting auto_error=False makes it return None instead, giving you control over the error. Used here to check header OR query param before deciding to reject.
Security() works like Depends() but also registers the security scheme in the OpenAPI spec. This makes the “Authorize” button appear in Swagger UI, allowing testers to set their API key once and use it on all protected endpoints.
JWT Authentication (OAuth2 + Bearer Tokens)
IntermediateJWT (JSON Web Token) is the industry standard for stateless authentication. The server creates a signed token containing user data; the client sends it with every request; the server verifies the signature without a database lookup. FastAPI uses OAuth2PasswordBearer to extract the token from the Authorization: Bearer TOKEN header.
# full JWT auth flow
from datetime import datetime, timedelta, timezone
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt # pip install python-jose[cryptography]
from pydantic import BaseModel
from typing import Annotated
app = FastAPI()
# ── Configuration ─────────────────────────────────────────────────────────────
SECRET_KEY = "replace-with-32-byte-random-hex-in-production" # openssl rand -hex 32
ALGORITHM = "HS256" # HMAC with SHA-256 — symmetric signing
EXPIRE_MINUTES = 30
# tokenUrl must match the path of your login endpoint
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
# ── Token schemas ─────────────────────────────────────────────────────────────
class Token(BaseModel):
access_token: str
token_type: str = "bearer"
# ── Create a JWT token ────────────────────────────────────────────────────────
def create_access_token(user_id: int, username: str) -> str:
payload = {
"sub": str(user_id), # "sub" (subject) = the user this token is for
"username": username,
"exp": datetime.now(timezone.utc) + timedelta(minutes=EXPIRE_MINUTES),
}
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
# ── Validate token and extract user (reusable dependency) ────────────────────
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)] # extracts Bearer token from header
) -> dict:
auth_error = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id = payload.get("sub")
if user_id is None:
raise auth_error
except JWTError: # expired, tampered, invalid signature
raise auth_error
# In real app: look up user in DB to check if still active
return {"user_id": int(user_id), "username": payload.get("username")}
# ── Login endpoint — issues the token ────────────────────────────────────────
@app.post("/auth/token", response_model=Token)
def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
# OAuth2 spec requires form data (not JSON) — form_data.username, .password
# In real app: verify hashed password from DB
if form_data.username != "alice" or form_data.password != "secret":
raise HTTPException(status_code=400, detail="Incorrect credentials")
token = create_access_token(user_id=1, username=form_data.username)
return Token(access_token=token)
# ── Protected route ───────────────────────────────────────────────────────────
@app.get("/users/me")
def get_me(current_user: Annotated[dict, Depends(get_current_user)]):
return current_user # {"user_id": 1, "username": "alice"}A security utility that looks for an Authorization: Bearer TOKEN header in each request and extracts the token string. If the header is missing, it raises 401 automatically. Also registers the scheme in OpenAPI so Swagger’s “Authorize” button appears.
encode: creates a signed token string from a dict payload using your secret key. decode: verifies the signature AND checks expiry, returning the payload dict. If the token was tampered with or expired, JWTError is raised.
The JWT “subject” claim — standard way to identify who the token is for. Should be a stable unique identifier (user ID from DB), not something that can change (like a username). Use the token’s sub to look up the user in your database.
A built-in FastAPI/Starlette class that reads username and password from form data (not JSON). The OAuth2 spec requires this for the token endpoint. Many developers expect JSON and get confused by 422 errors — the form data requirement is in the OAuth2 spec.
The server never stores the token. It creates a signed token and trusts it later because only the server knows the SECRET_KEY. A tampered token will fail signature verification. This is what makes JWTs scalable — no session lookup in Redis or DB on every request.
⚠ Gotcha — JWT tokens cannot be revoked without a blacklist
Once issued, a JWT is valid until it expires — you can’t “log out” a user by deleting anything server-side. For token revocation (e.g., on password change, account ban), you need a server-side token blocklist in Redis (store the jti claim of revoked tokens). This partially re-adds statefulness, so consider short expiry times + refresh tokens.
SQLAlchemy — Sync & Async
IntermediateSQLAlchemy is the standard ORM for FastAPI projects. For async routes (the norm in FastAPI), you use SQLAlchemy’s async engine with asyncpg or aiosqlite. The key pattern: a yield dependency provides a database session per request and closes it cleanly after.
# async SQLAlchemy setup
from sqlalchemy.ext.asyncio import (
create_async_engine, AsyncSession, async_sessionmaker
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Integer, select
from fastapi import FastAPI, Depends, HTTPException
# ── Engine & session factory ──────────────────────────────────────────────────
# postgresql+asyncpg → async PostgreSQL driver
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"
engine = create_async_engine(
DATABASE_URL,
echo=False, # set True to log all SQL (development only)
pool_size=10, # connections in pool
pool_pre_ping=True, # test connections before use (detect stale connections)
)
# expire_on_commit=False: don't expire attributes after commit
# Without this, accessing model.id after commit raises MissingGreenlet in async
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
# ── Models using SQLAlchemy 2.0 typed syntax ─────────────────────────────────
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
# ── Per-request session dependency ───────────────────────────────────────────
async def get_db():
async with AsyncSessionLocal() as session: # opens session
yield session # route handler uses session here
# session is closed automatically when 'async with' block exits
# ── Routes using async DB ─────────────────────────────────────────────────────
app = FastAPI()
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none() # None if not found
if not user:
raise HTTPException(status_code=404, detail="User not found")
return {"id": user.id, "name": user.name, "email": user.email}
@app.post("/users/", status_code=201)
async def create_user(name: str, email: str, db: AsyncSession = Depends(get_db)):
user = User(name=name, email=email)
db.add(user)
await db.commit()
await db.refresh(user) # reloads user from DB to get generated id
return {"id": user.id, "name": user.name}By default, SQLAlchemy expires all model attributes after a commit (marks them stale, requiring a DB query to reload). In async code, this lazy reload causes a MissingGreenlet error. Setting expire_on_commit=False keeps attribute values in memory after commit.
SQLAlchemy 2.0’s typed column syntax. Mapped[int] = not nullable; Mapped[int | None] = nullable. This replaces the old Column(Integer, nullable=False) syntax and gives full type-checker support — your IDE knows user.id is always an int.
Extracts the single scalar result from a SQLAlchemy Result. Returns the object if found, None if not found, raises if multiple rows returned. Use scalars().all() for lists.
After db.add(user) + await db.commit(), the user object doesn’t have its auto-generated id yet (it was generated by the DB during the INSERT). refresh() re-fetches the row from the DB, populating all generated fields.
💬 Interview Tip
Common interview question: “What is N+1 query problem and how do you fix it?” — If you load a list of users and then access user.orders for each one, SQLAlchemy issues N+1 queries (1 for users + N for each user’s orders). Fix with select(User).options(selectinload(User.orders)) — loads all orders in a single second query.
Alembic Migrations
IntermediateAlembic is SQLAlchemy’s database migration tool. It tracks schema changes (adding columns, creating tables) in versioned Python scripts that can be applied or rolled back. Never use create_all() in production — it can’t handle incremental changes. Alembic is the proper way to evolve your schema over time.
# 1. Install and initialize
pip install alembic
alembic init alembic # creates alembic/ directory and alembic.ini
# 2. Edit alembic/env.py (see below)
# 3. Generate migration from your model changes
alembic revision --autogenerate -m "add users table"
# Creates: alembic/versions/abc123_add_users_table.py
# 4. Review the generated migration (always check before running!)
# Then apply:
alembic upgrade head # apply all pending migrations
# 5. Other useful commands:
alembic upgrade +1 # apply one migration forward
alembic downgrade -1 # undo last migration
alembic downgrade base # roll back ALL migrations
alembic current # show current schema version
alembic history # show migration history# alembic/env.py (key section)
from app.database import Base # your DeclarativeBase
from app import models # MUST import all model files so Base "sees" them
# If models aren't imported here, Alembic doesn't know about the tables
# and will generate migrations that DROP all your tables!
target_metadata = Base.metadata # tells Alembic what tables should exist
# In run_migrations_online():
# Use NullPool for migrations — no connection reuse, cleaner for scripts
from sqlalchemy.pool import NullPool
connectable = engine_from_config(..., poolclass=NullPool)Alembic compares your SQLAlchemy models (what your code says the schema should be) against the actual database (what it currently is) and generates a migration that transforms one into the other. Always review the generated script — autogenerate isn’t perfect.
upgrade head applies all pending migrations up to the latest. downgrade -1 runs the downgrade() function of the last applied migration — reverting its changes. This is your rollback mechanism if a deployment goes wrong.
Alembic builds the “target metadata” from what’s registered on Base.metadata. A model only registers itself when its module is imported. If you forget to import app.models.user, Alembic thinks the users table shouldn’t exist and generates a DROP TABLE migration.
Environment Config with Pydantic Settings
Intermediatepydantic-settings provides BaseSettings, which automatically reads configuration from environment variables and .env files, validates types, and gives you a clean Python object. It’s the canonical way to manage config in FastAPI — much better than raw os.environ.get() calls scattered through your code.
# app/config.py
# pip install pydantic-settings
from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import PostgresDsn, RedisDsn, SecretStr
from functools import lru_cache
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env", # read variables from .env file
env_file_encoding="utf-8",
case_sensitive=False, # DATABASE_URL == database_url
extra="ignore", # ignore unknown env vars silently
)
# App settings
app_name: str = "My API"
debug: bool = False
# Database — PostgresDsn validates it's a valid Postgres URL
database_url: PostgresDsn
db_pool_size: int = 10
# Redis (optional)
redis_url: RedisDsn | None = None
# Security — SecretStr hides the value in logs and repr()
secret_key: SecretStr
access_token_expire_minutes: int = 30
@property
def secret_key_str(self) -> str:
return self.secret_key.get_secret_value() # get the actual value
# lru_cache ensures Settings() is only created ONCE per process
# Without it, .env would be re-read on every request
@lru_cache
def get_settings() -> Settings:
return Settings()
# ── Usage in routes and dependencies ─────────────────────────────────────────
from fastapi import Depends
from typing import Annotated
@app.get("/info")
def app_info(settings: Annotated[Settings, Depends(get_settings)]):
return {
"app": settings.app_name,
"debug": settings.debug,
# Never return settings.secret_key — it would be in the response!
}Like BaseModel but reads field values from environment variables first (falling back to .env file, then defaults). Env vars override .env which overrides defaults — perfect for the 12-factor app pattern where production config comes from the environment.
A special Pydantic type that wraps a string and hides its value. repr(settings) shows secret_key=SecretStr('**********') instead of the actual value. This prevents secrets from appearing in log files or error messages. Access the real value with .get_secret_value().
Without lru_cache, every call to Depends(get_settings) would create a new Settings instance and re-read the .env file. With it, the first call creates the instance and all subsequent calls return the same cached object — essentially a singleton.
Pydantic’s specialized URL types that validate format at startup. If DATABASE_URL is set to a malformed string, your app will fail to start with a clear error — instead of failing silently at runtime when the first request tries to connect to the database.
Testing with pytest & TestClient / httpx
IntermediateFastAPI’s TestClient wraps your app in a mock HTTP server so you can test endpoints without running a real server. The most important FastAPI-specific testing feature is app.dependency_overrides — it lets you swap real dependencies (database, auth) with test fakes cleanly, without monkey-patching.
# tests/test_items.py
import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient, ASGITransport
from app.main import app
from app.dependencies import get_db, get_current_user
# ── Sync TestClient — works for most tests ────────────────────────────────────
client = TestClient(app)
def test_root():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, FastAPI!"}
def test_create_item_success():
response = client.post("/items/", json={"name": "Widget", "price": 9.99})
assert response.status_code == 201
assert response.json()["name"] == "Widget"
def test_create_item_missing_field():
response = client.post("/items/", json={"name": "Widget"}) # missing price
assert response.status_code == 422 # Pydantic validation error
assert "price" in str(response.json())
# ── Dependency overrides — swap real deps with test fakes ────────────────────
def fake_get_db():
"""Returns an in-memory test database instead of real DB."""
test_db = {"users": {1: {"id": 1, "name": "Test User"}}}
yield test_db
def fake_get_current_user():
"""Returns a fake user — no JWT validation needed in tests."""
return {"user_id": 1, "username": "testuser"}
@pytest.fixture(autouse=True) # runs before/after every test
def override_deps():
app.dependency_overrides[get_db] = fake_get_db
app.dependency_overrides[get_current_user] = fake_get_current_user
yield
app.dependency_overrides.clear() # ALWAYS reset after test
# ── Async testing with httpx (for async routes) ───────────────────────────────
@pytest.mark.asyncio
async def test_async_route():
# ASGITransport lets httpx talk to the app directly (no real HTTP server)
async with AsyncClient(
transport=ASGITransport(app=app),
base_url="http://test"
) as ac:
response = await ac.get("/users/me")
assert response.status_code == 200
# ── Test auth flow ─────────────────────────────────────────────────────────────
def test_protected_with_token():
# First get a token
login = client.post("/auth/token", data={"username": "alice", "password": "secret"})
assert login.status_code == 200
token = login.json()["access_token"]
# Then use it
response = client.get("/users/me", headers={"Authorization": f"Bearer {token}"})
assert response.status_code == 200Creates a synchronous HTTP client backed by your FastAPI app. Requests go directly to the ASGI app (no real network). Based on requests library, so the API is familiar: client.get(url), client.post(url, json={...}).
A dict mapping real dependency functions → test replacement functions. When FastAPI resolves dependencies for a request, it checks this dict first. This is the official way to mock dependencies — much cleaner than unittest.mock.patch() on internal functions.
Removes all overrides, restoring real dependencies. Critical: if you forget this in teardown, overrides bleed into other tests. Use a pytest fixture with yield and .clear() in the teardown to guarantee cleanup.
Lets httpx.AsyncClient send requests directly to an ASGI app without a running server. Required for async tests. The base_url="http://test" is a dummy URL — the transport ignores it and routes to your app directly.
In TestClient/httpx: json={...} sends application/json; data={...} sends application/x-www-form-urlencoded. The OAuth2 token endpoint requires data= (form data), which is why login tests use data= not json=.
💬 Interview Tip
“How do you test a FastAPI route that requires authentication?” — Two approaches: (1) use app.dependency_overrides[get_current_user] = lambda: test_user to bypass auth entirely; or (2) call the login endpoint first, extract the token, and pass it as a header. Option 1 is faster for unit testing routes; option 2 is better for integration testing the full auth flow.