FastAPI is not a thin wrapper around Starlette. The combination of Pydantic v2 for validation, Depends() for dependency injection, and async-native request handling makes it the closest Python equivalent to a typed HTTP framework. The original version of this post had a broken handler signature and no production content. This is the rebuild.
Installation#
pip install fastapi "uvicorn[standard]"The [standard] extra installs uvloop (faster event loop), httptools (faster HTTP parsing), and websockets. Always use it.
Pydantic Models: Request, Response, and Validation#
from datetime import datetime
from typing import Annotated
from pydantic import BaseModel, Field, field_validator
class CreateUserRequest(BaseModel):
username: Annotated[str, Field(min_length=3, max_length=50, pattern=r"^[a-z0-9_]+$")]
email: str
age: Annotated[int, Field(ge=18, le=120)]
@field_validator("email")
@classmethod
def email_must_have_at(cls, v: str) -> str:
if "@" not in v:
raise ValueError("invalid email address")
return v.lower()
class UserResponse(BaseModel):
id: int
username: str
email: str
created_at: datetime
model_config = {"from_attributes": True} # enables ORM modePydantic v2 validates on construction. If age is a string that cannot be coerced to an integer, FastAPI returns a 422 with a structured error body – no manual validation code required.
Path Parameters, Query Parameters, and Request Body#
from fastapi import FastAPI, Query
from models import CreateUserRequest, UserResponse
app = FastAPI(title="User API", version="1.0.0")
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
user_id: int,
include_deleted: bool = Query(default=False, description="include soft-deleted users"),
):
# user_id is a path parameter (required, typed as int)
# include_deleted is a query parameter with a default
user = await db.get_user(user_id, include_deleted=include_deleted)
if user is None:
raise HTTPException(status_code=404, detail="user not found")
return user
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(body: CreateUserRequest):
# body is automatically parsed from the JSON request body
user = await db.create_user(body.username, body.email, body.age)
return userresponse_model=UserResponse is critical. It tells FastAPI to serialize the return value through the Pydantic model, which strips any fields not declared in UserResponse. Without it, internal database fields (password hashes, internal IDs, flags) can leak to API consumers.
Async Endpoints: When to Use async def vs def#
import httpx
import time
# Use async def for I/O-bound work: database queries, HTTP calls, file reads
@app.get("/async-example")
async def fetch_remote():
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.example.com/data")
return resp.json()
# Use def (sync) for CPU-bound work: image processing, heavy computation
# FastAPI runs sync handlers in a threadpool automatically
@app.get("/sync-example")
def compute_heavy():
# This runs in a threadpool, not blocking the event loop
result = expensive_cpu_computation()
return {"result": result}Do not call blocking functions (time.sleep, synchronous database drivers, requests.get) inside async def handlers. They block the event loop and serialize all requests. Use await asyncio.sleep, async database drivers (asyncpg, motor), and httpx.AsyncClient.
Dependency Injection with Depends()#
Dependencies are the cleanest way to share authentication, database sessions, and configuration across endpoints.
from fastapi import Depends, HTTPException, Header
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
# --- Database session dependency ---
async_session_factory: async_sessionmaker[AsyncSession] | None = None
async def get_db() -> AsyncSession:
async with async_session_factory() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
# --- JWT Bearer authentication dependency ---
async def get_current_user(
authorization: str = Header(),
db: AsyncSession = Depends(get_db),
) -> dict:
if not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="missing bearer token")
token = authorization.removeprefix("Bearer ")
payload = verify_jwt(token) # raises on invalid token
if payload is None:
raise HTTPException(status_code=401, detail="invalid token")
user = await db.get(User, payload["sub"])
if user is None:
raise HTTPException(status_code=401, detail="user not found")
return userfrom fastapi import Depends
from dependencies import get_db, get_current_user
@app.get("/me", response_model=UserResponse)
async def get_profile(current_user=Depends(get_current_user)):
return current_user
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(
user_id: int,
db: AsyncSession = Depends(get_db),
_: dict = Depends(get_current_user), # auth check without using the value
):
await db.delete(await db.get(User, user_id))FastAPI resolves the dependency graph automatically. get_current_user itself depends on get_db, so one database session is shared across both.
HTTPException and Custom Exception Handlers#
from fastapi import Request
from fastapi.responses import JSONResponse
class RateLimitExceeded(Exception):
def __init__(self, retry_after: int):
self.retry_after = retry_after
@app.exception_handler(RateLimitExceeded)
async def rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONResponse:
return JSONResponse(
status_code=429,
content={"detail": "rate limit exceeded", "retry_after": exc.retry_after},
headers={"Retry-After": str(exc.retry_after)},
)Background Tasks#
Use BackgroundTasks for work that should not block the HTTP response: sending emails, recording analytics, triggering async pipelines.
from fastapi import BackgroundTasks
async def send_welcome_email(email: str, username: str) -> None:
# runs after the response is sent
await email_client.send(
to=email,
subject="Welcome",
body=f"Hi {username}, your account is ready.",
)
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(body: CreateUserRequest, background_tasks: BackgroundTasks):
user = await db.create_user(body.username, body.email)
background_tasks.add_task(send_welcome_email, body.email, body.username)
return user # response sent immediately; email runs afterBackgroundTasks runs in the same process after the response is sent. For durable background work (retries, persistence across restarts), use a proper task queue like Celery or ARQ.
Production Deployment#
# Single process, auto-reload on file change
uvicorn main:app --reload --host 0.0.0.0 --port 8000Never use --reload in production. It watches the filesystem and restarts the process – it is not safe or performant for production traffic.
# Multiple workers for CPU parallelism
# Rule of thumb: (2 * CPU_count) + 1
gunicorn main:app \
--workers 5 \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000 \
--timeout 30 \
--graceful-timeout 10 \
--access-logfile -Each Gunicorn worker is a separate process running a Uvicorn event loop. This gives you process-level parallelism (Gunicorn) and non-blocking I/O (Uvicorn) in combination.
For containerized deployments, use this as your CMD in a Dockerfile:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
ENV WORKERS=4
CMD gunicorn main:app \
--workers $WORKERS \
--worker-class uvicorn.workers.UvicornWorker \
--bind 0.0.0.0:8000Load environment configuration via Pydantic Settings:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
environment: str = "production"
debug: bool = False
model_config = {"env_file": ".env"}
settings = Settings()Common mistakes and how to avoid them
Using def instead of async def for I/O-bound handlers
A synchronous database call inside an async def handler blocks the event loop while waiting for the database. Every other request queues behind it. Use async drivers (asyncpg, motor, aioredis) with await.
Not using response_model
Without response_model, FastAPI serializes the raw return value. If your handler returns a SQLAlchemy ORM object, internal fields (hashed passwords, internal flags, foreign keys) will appear in the response. Always declare response_model on every endpoint.
No dependency injection for database sessions
Creating a new database connection per handler without a dependency leads to connection leaks when exceptions occur before the connection is closed. The get_db dependency with yield ensures the session is always rolled back and closed, even on exceptions.
Running the Uvicorn dev server in production
uvicorn main:app starts a single-process server with no worker management. A single uncaught exception takes down the entire service. Gunicorn manages worker restarts, graceful reloads, and process supervision.
Broken handler signatures
The @app.get("/") handler does not take request and response as positional arguments. FastAPI uses function signature inspection to inject path parameters, query parameters, headers, and bodies. The correct signature for a simple GET with no parameters is async def handler() -> dict.
If you want to go deeper on any of this, I offer 1:1 coaching sessions for engineers working on AI integration, cloud architecture, and platform engineering. Book a session (50 EUR / 60 min) or reach out at manuel.fedele+website@gmail.com.
