FastAPI Deep Dive

0/4 in this phase0/54 across the roadmap

📖 Concept

FastAPI is a modern, high-performance Python web framework built on top of Starlette (ASGI toolkit) and Pydantic (data validation). It has rapidly become the most popular choice for building Python APIs due to its speed, developer experience, and automatic documentation.

Core Concepts:

Pydantic Models are the backbone of FastAPI. Every request body, response, and query parameter can be defined as a Pydantic model with type hints, validators, and default values. FastAPI uses these models to automatically parse, validate, serialize, and document your API.

Dependency Injection is FastAPI's most powerful feature. The `Depends()` function lets you declare dependencies that are resolved at request time. Dependencies can depend on other dependencies (forming a graph), they can be sync or async, and they handle cleanup via `yield`. Common uses: database sessions, authentication, rate limiting, pagination.

Middleware intercepts every request/response cycle. FastAPI supports both Starlette-style middleware (classes with `dispatch` method) and pure ASGI middleware. Use cases: CORS, request timing, logging, authentication headers, compression.

Background Tasks let you run functions after returning a response. FastAPI's `BackgroundTasks` parameter queues work to be executed after the response is sent — ideal for sending emails, logging analytics, or cleanup operations without blocking the client.

Async Handlers are where FastAPI shines. Handlers declared with `async def` run on the async event loop, enabling non-blocking I/O. Handlers declared with plain `def` are automatically run in a thread pool. The rule: use `async def` when calling `await`-able code (async DB drivers, httpx), use `def` for synchronous blocking code.

Key architectural decisions:

  • Use `APIRouter` to organize endpoints into modules (like Flask Blueprints)
  • Use `Lifespan` events for startup/shutdown logic (DB connection pools, ML model loading)
  • Use response models (`response_model`) to control what gets serialized — never expose internal fields

💻 Code Example

codeTap to expand ⛶
1# ============================================================
2# FastAPI Production Patterns
3# ============================================================
4# from fastapi import (
5# FastAPI, Depends, HTTPException, BackgroundTasks,
6# Query, Path, status, Request
7# )
8# from fastapi.middleware.cors import CORSMiddleware
9# from pydantic import BaseModel, Field, field_validator, ConfigDict
10# from contextlib import asynccontextmanager
11# from typing import Annotated
12# import asyncio
13# import time
14# import logging
15#
16# logger = logging.getLogger(__name__)
17#
18#
19# # ============================================================
20# # Pydantic Models — request/response validation
21# # ============================================================
22# class UserCreate(BaseModel):
23# """Request model with validation rules."""
24# model_config = ConfigDict(
25# json_schema_extra={
26# "examples": [{"username": "janedoe", "email": "jane@example.com"}]
27# }
28# )
29#
30# username: str = Field(..., min_length=3, max_length=30, pattern=r"^[a-z0-9_]+$")
31# email: str = Field(..., max_length=255)
32# full_name: str | None = Field(default=None, max_length=100)
33#
34# @field_validator("email")
35# @classmethod
36# def validate_email(cls, v):
37# if "@" not in v or "." not in v.split("@")[-1]:
38# raise ValueError("Invalid email format")
39# return v.lower()
40#
41#
42# class UserResponse(BaseModel):
43# """Response model — controls what the client sees."""
44# id: int
45# username: str
46# email: str
47# full_name: str | None = None
48# # NOTE: password hash is NOT included in the response model
49#
50#
51# # ============================================================
52# # Dependency Injection — composable request dependencies
53# # ============================================================
54#
55# # BAD: Hardcoded database connection in every handler
56# # @app.get("/users")
57# # async def get_users():
58# # db = sqlite3.connect("app.db") # created every request!
59# # users = db.execute("SELECT * FROM users").fetchall()
60# # db.close()
61# # return users
62#
63# # GOOD: Dependency injection with yield (setup + teardown)
64# async def get_db():
65# """Database session dependency with automatic cleanup."""
66# db = AsyncSession(engine)
67# try:
68# yield db # value injected into handler
69# finally:
70# await db.close() # teardown runs after response
71#
72#
73# async def get_current_user(
74# request: Request,
75# db: AsyncSession = Depends(get_db),
76# ):
77# """Auth dependency — depends on get_db (dependency chain)."""
78# token = request.headers.get("Authorization", "").replace("Bearer ", "")
79# if not token:
80# raise HTTPException(status_code=401, detail="Not authenticated")
81# user = await db.execute(select(User).where(User.token == token))
82# if not user:
83# raise HTTPException(status_code=401, detail="Invalid token")
84# return user.scalar_one()
85#
86#
87# class PaginationParams:
88# """Reusable pagination dependency as a class."""
89# def __init__(
90# self,
91# page: int = Query(1, ge=1, description="Page number"),
92# per_page: int = Query(20, ge=1, le=100, description="Items per page"),
93# ):
94# self.offset = (page - 1) * per_page
95# self.limit = per_page
96#
97#
98# # ============================================================
99# # Lifespan — startup/shutdown events
100# # ============================================================
101# @asynccontextmanager
102# async def lifespan(app: FastAPI):
103# """Replaces deprecated @app.on_event startup/shutdown."""
104# logger.info("Starting up: initializing DB pool...")
105# # app.state.db_pool = await create_pool(...)
106# yield
107# logger.info("Shutting down: closing DB pool...")
108# # await app.state.db_pool.close()
109#
110#
111# app = FastAPI(title="User API", version="2.0", lifespan=lifespan)
112#
113# # Middleware
114# app.add_middleware(
115# CORSMiddleware,
116# allow_origins=["https://example.com"],
117# allow_methods=["*"],
118# allow_headers=["*"],
119# )
120#
121#
122# # ============================================================
123# # Background Tasks — fire-and-forget after response
124# # ============================================================
125# async def send_welcome_email(email: str, username: str):
126# """Runs AFTER the response is sent to the client."""
127# await asyncio.sleep(2) # simulate email sending
128# logger.info(f"Welcome email sent to {email}")
129#
130#
131# @app.post("/users", response_model=UserResponse, status_code=201)
132# async def create_user(
133# user: UserCreate,
134# background_tasks: BackgroundTasks,
135# db: AsyncSession = Depends(get_db),
136# ):
137# # Pydantic already validated the input
138# db_user = User(**user.model_dump())
139# db.add(db_user)
140# await db.commit()
141# await db.refresh(db_user)
142#
143# # Schedule background work — does NOT block the response
144# background_tasks.add_task(send_welcome_email, user.email, user.username)
145#
146# return db_user
147#
148#
149# @app.get("/users", response_model=list[UserResponse])
150# async def list_users(
151# pagination: PaginationParams = Depends(),
152# current_user: User = Depends(get_current_user),
153# db: AsyncSession = Depends(get_db),
154# ):
155# result = await db.execute(
156# select(User).offset(pagination.offset).limit(pagination.limit)
157# )
158# return result.scalars().all()

🏋️ Practice Exercise

Exercises:

  1. Build a FastAPI application with Pydantic models for a `Product` resource. Include `field_validator` rules: price must be positive, SKU must match a regex pattern, and description must be between 10 and 1000 characters. Test that invalid inputs return 422 with clear error messages.

  2. Implement a dependency injection chain: `get_db()` yields a database session, `get_current_user()` depends on `get_db()` and extracts the user from a JWT token, and `require_admin()` depends on `get_current_user()` and checks for admin role. Apply `require_admin` to a DELETE endpoint.

  3. Add custom middleware that logs the request method, path, status code, and response time for every request. Store the logs in a list and expose a `GET /metrics` endpoint that returns the last 100 entries.

  4. Create a background task system: when a user signs up via `POST /register`, immediately return 201, then send a welcome email and generate a default avatar image in the background. Verify the tasks complete using a `GET /tasks/{task_id}` status endpoint.

  5. Organize a FastAPI app into multiple `APIRouter` modules: `users.py`, `products.py`, and `orders.py`. Each router should have its own prefix, tags, and shared dependencies. Wire them together in `main.py` and verify the auto-generated docs group endpoints by tag.

  6. Implement rate limiting as a FastAPI dependency. The dependency should track requests per IP using an in-memory dictionary and raise `HTTPException(429)` if the limit is exceeded. Add a sliding window algorithm instead of a simple counter.

⚠️ Common Mistakes

  • Declaring handlers as async def but calling synchronous blocking code inside them (e.g., requests.get(), time.sleep(), synchronous ORM queries). This blocks the entire event loop. Either use def (runs in thread pool) or use async libraries (httpx, asyncpg).

  • Not using response_model on endpoints, which means internal fields (password hashes, internal IDs, soft-delete flags) leak to the client. Always define a separate response model that excludes sensitive data.

  • Creating a new database connection inside every handler instead of using Depends() with a yielding dependency. This wastes resources and makes testing impossible — dependencies can be overridden in tests with app.dependency_overrides.

  • Putting all endpoints in a single main.py file instead of using APIRouter for modular organization. This creates a monolith that is hard to maintain and test. Split by domain: routers/users.py, routers/products.py, etc.

  • Using @app.on_event('startup') and @app.on_event('shutdown') which are deprecated. Use the lifespan async context manager instead, which provides cleaner resource management and is the officially supported pattern.

💼 Interview Questions

🎤 Mock Interview

Practice a live interview for FastAPI Deep Dive