FastAPI Deep Dive
📖 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
1# ============================================================2# FastAPI Production Patterns3# ============================================================4# from fastapi import (5# FastAPI, Depends, HTTPException, BackgroundTasks,6# Query, Path, status, Request7# )8# from fastapi.middleware.cors import CORSMiddleware9# from pydantic import BaseModel, Field, field_validator, ConfigDict10# from contextlib import asynccontextmanager11# from typing import Annotated12# import asyncio13# import time14# import logging15#16# logger = logging.getLogger(__name__)17#18#19# # ============================================================20# # Pydantic Models — request/response validation21# # ============================================================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# @classmethod36# 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: int45# username: str46# email: str47# full_name: str | None = None48# # NOTE: password hash is NOT included in the response model49#50#51# # ============================================================52# # Dependency Injection — composable request dependencies53# # ============================================================54#55# # BAD: Hardcoded database connection in every handler56# # @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 users62#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 handler69# finally:70# await db.close() # teardown runs after response71#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_page95# self.limit = per_page96#97#98# # ============================================================99# # Lifespan — startup/shutdown events100# # ============================================================101# @asynccontextmanager102# 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# yield107# 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# # Middleware114# 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 response124# # ============================================================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 sending128# 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 input138# 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 response144# background_tasks.add_task(send_welcome_email, user.email, user.username)145#146# return db_user147#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:
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.
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.
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.
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.
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.
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 defbut calling synchronous blocking code inside them (e.g.,requests.get(),time.sleep(), synchronous ORM queries). This blocks the entire event loop. Either usedef(runs in thread pool) or use async libraries (httpx,asyncpg).Not using
response_modelon 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 withapp.dependency_overrides.Putting all endpoints in a single
main.pyfile instead of usingAPIRouterfor 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 thelifespanasync 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