FastAPI

FastAPI is a modern Python web framework for building APIs. It uses Python type hints to do request validation, serialization, and even auto-generate interactive docs. In simple language: we write a normal Python function with type hints, and FastAPI figures out the rest.

It sits on top of two libraries:

• Starlette — handles the async HTTP plumbing (routing, middleware, websockets).
• Pydantic — handles the data validation using type hints.

So FastAPI is really a thin, opinionated layer that glues these two together and adds OpenAPI/Swagger on top.

Why people love it

Think of it like Flask, but with three big upgrades:

1. Async-first. We can async def any endpoint, hit databases or other APIs concurrently, and the event loop doesn’t block.
2. Validation comes free. Type-hint a parameter as int and FastAPI rejects strings with a clean 422 error. No more manual request.json.get() checks.
3. Docs come free. Go to /docs and there’s a full Swagger UI. /redoc gives the ReDoc version. Both auto-generated from our function signatures.

Quick example

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    in_stock: bool = True

@app.post("/items/")
async def create_item(item: Item):
    return {"item": item, "tax": item.price * 0.18}

That’s a real endpoint. It validates the body, serializes the response, and shows up in /docs with a try-it-out button. We didn’t write a single line of validation code.

When to reach for it

• Building a REST or GraphQL API where types matter (data ingestion, ML model serving, internal services).
• Need async I/O — calling many DB queries, third-party APIs, or websockets.
• Want OpenAPI specs without hand-writing YAML.

When not to

• Building a server-rendered website with templates and forms. Django is better there.
• Tiny script that just needs one endpoint. Flask or even http.server is enough.

The interview pitch

If someone asks “what is FastAPI” in a screening round, the one-liner is: “FastAPI is a modern, async Python framework built on Starlette and Pydantic that uses type hints for automatic request validation, response serialization, and OpenAPI doc generation.” That sentence covers everything an interviewer wants to hear.

This is the classic “tell me the difference between…” interview question for Python backend roles. The short answer: Django is a full-stack batteries-included framework, Flask is a minimal sync micro-framework, and FastAPI is a modern async API framework with type-driven validation.

Let’s break down the actual differences.

Async-first vs sync

Flask and Django were built in a sync world. Each request occupies a worker thread or process from start to finish. If our handler hits a slow database query, that worker just sits there waiting.

FastAPI is async by default. We write async def and use await for I/O. One worker can handle hundreds of concurrent requests because while one waits on the DB, another runs.

# FastAPI - async handler
@app.get("/orders/{user_id}")
async def get_orders(user_id: int):
    user = await db.fetch_user(user_id)
    orders = await db.fetch_orders(user_id)
    return {"user": user, "orders": orders}

# Flask - sync handler
@app.route("/orders/<int:user_id>")
def get_orders(user_id):
    user = db.fetch_user(user_id)
    orders = db.fetch_orders(user_id)
    return {"user": user, "orders": orders}

Django supports async views since 3.1, but a lot of the ecosystem (ORM, middleware) is still sync-leaning.

Validation

In Flask we pull values off request.json and validate them ourselves (or pull in Marshmallow). In Django REST Framework we write a Serializer class. In FastAPI it’s just a type hint.

# FastAPI
@app.post("/users/")
async def create_user(name: str, age: int):
    return {"name": name, "age": age}

That age: int automatically rejects "twenty" with a 422 response. No extra code.

Batteries vs choose-your-own

Django gives us ORM, admin, auth, sessions, migrations, templates — all in one box. Great for content sites, CRUD apps, internal tools.

FastAPI gives us routing, validation, docs. We bring our own ORM (usually SQLAlchemy or Tortoise), auth library, etc. More flexible, more decisions to make.

Flask sits in between but leans toward minimal — bring everything yourself.

Which to pick?

• Django — content site, lots of CRUD, want admin panel for free, team likes monoliths.
• Flask — small sync API, lots of legacy code, simple internal tools.
• FastAPI — modern API, async I/O, ML model serving, microservices, want OpenAPI for free.

The interview answer

“FastAPI is async-first with built-in Pydantic validation and auto OpenAPI docs, Flask is a minimal sync micro-framework, and Django is a batteries-included full-stack framework with ORM and admin. For new high-throughput APIs I’d reach for FastAPI.” That covers 90% of the question.

WSGI (Web Server Gateway Interface) and ASGI (Asynchronous Server Gateway Interface) are both contracts between a Python web framework and the server that runs it. In simple language: they’re the standard way a server (gunicorn, uvicorn) and a framework (Flask, FastAPI) talk to each other.

The only difference is WSGI is sync, ASGI is async.

What WSGI looks like

WSGI was defined in PEP 3333 back in 2010. The contract is dead simple: the server hands the framework an environ dict and a start_response callable, and the framework returns an iterable of bytes.

# Minimal WSGI app
def app(environ, start_response):
    start_response("200 OK", [("Content-Type", "text/plain")])
    return [b"Hello WSGI"]

One request, one function call, one response. The function is sync — it returns when the response is ready. Flask and Django are WSGI frameworks. Gunicorn is a WSGI server.

The catch: there’s no way to handle long-lived connections (websockets, server-sent events) or to suspend a handler mid-execution. The interface assumes “request in, response out, done.”

What ASGI looks like

ASGI is the async successor. The contract is an async callable that receives scope, receive, and send. Now the handler can await things and yield back to the event loop.

# Minimal ASGI app
async def app(scope, receive, send):
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"text/plain")],
    })
    await send({
        "type": "http.response.body",
        "body": b"Hello ASGI",
    })

Because send and receive are awaitable, the same protocol works for HTTP, websockets, and background events. FastAPI and Starlette are ASGI frameworks. Uvicorn and Hypercorn are ASGI servers.

Why FastAPI needs uvicorn (or similar)

FastAPI emits ASGI. Gunicorn (the standard WSGI server) can’t speak ASGI on its own. We need an ASGI server like uvicorn:

uvicorn main:app --host 0.0.0.0 --port 8000

For dev, uvicorn alone is fine. For prod, we want process management — multiple workers, graceful reloads, OS signal handling. That’s where gunicorn comes back in.

gunicorn + uvicorn workers

Gunicorn is a great process manager (forking, worker lifecycle, signals) but it’s WSGI by default. The trick: it supports plug-in worker classes, and uvicorn ships one. So we run:

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

What this means in simple language: gunicorn is the boss process that forks 4 child workers. Each child is actually a uvicorn event loop. Gunicorn handles restarts and signals, uvicorn handles the ASGI requests. Best of both worlds.

Modern FastAPI deploys are increasingly skipping gunicorn entirely and running uvicorn directly with --workers 4, since uvicorn now has its own multi-worker support. Both patterns work.

Interview cheat

If asked “why can’t we run FastAPI under gunicorn directly?” — because gunicorn’s default workers are WSGI, and FastAPI is ASGI. We either use -k uvicorn.workers.UvicornWorker or run uvicorn standalone.

Let’s get a FastAPI app actually running. Three steps: install, write a file, run uvicorn.

Install

pip install "fastapi[standard]"

The [standard] extra pulls in uvicorn and a few useful goodies (httpx for the test client, email-validator, etc.). If we want bare-bones we can do pip install fastapi uvicorn instead.

The smallest app

Create main.py:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello from FastAPI"}

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id, "name": f"Item #{item_id}"}

A few things to notice:

• app = FastAPI() creates the application instance. Everything attaches to this.
• @app.get("/") registers a GET handler at /. Same pattern as Flask but on the app instance.
• The function can be async def or plain def. FastAPI handles both — plain def runs in a threadpool so we don’t block the event loop.
• The dict we return gets serialized to JSON automatically.

Run it

uvicorn main:app --reload

Breaking down main:app — main is the Python module (the file main.py), app is the variable name inside that file. --reload watches files and restarts on change. Only use it in dev.

Output should look like:

INFO:     Uvicorn running on http://127.0.0.1:8000
INFO:     Started reloader process
INFO:     Application startup complete.

Hit http://localhost:8000/ in a browser and we get {"message": "Hello from FastAPI"}.

The free docs

This is the magic. Without writing any extra code, go to:

• http://localhost:8000/docs — interactive Swagger UI. We can expand each endpoint and hit “Try it out” to fire real requests.
• http://localhost:8000/redoc — ReDoc-style docs (cleaner, three-column layout, better for reading).
• http://localhost:8000/openapi.json — the raw OpenAPI 3.1 spec as JSON. Useful for client codegen.

Test it from the terminal

curl http://localhost:8000/items/42
# {"item_id":42,"name":"Item #42"}

curl http://localhost:8000/items/abc
# {"detail":[{"type":"int_parsing","loc":["path","item_id"],...}]}

That second response is FastAPI rejecting abc because we typed item_id as int. We got validation for free.

App-level options

FastAPI() accepts a bunch of kwargs that show up in the docs:

app = FastAPI(
    title="Inventory Service",
    description="Manages items, stock levels, and reorders.",
    version="0.1.0",
    docs_url="/swagger",     # change default /docs path
    redoc_url=None,          # disable redoc
)

In prod we usually set docs_url=None and redoc_url=None to hide internal docs from the public.

fastapi dev (newer way)

Recent FastAPI versions ship a CLI:

fastapi dev main.py

Same thing as uvicorn main:app --reload but with nicer output. For prod use fastapi run main.py which skips the reloader.

That’s it — we have a working API, free docs, and validation. The rest is just adding more routes.

A path operation in FastAPI lingo is just “a route plus an HTTP method.” Things like GET /users/{id} or POST /orders. We register them with decorators on the app instance.

The decorators

One per HTTP verb:

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def list_items():
    return [{"id": 1, "name": "Notebook"}]

@app.post("/items/")
async def create_item():
    return {"id": 2, "name": "Pen"}

@app.put("/items/{item_id}")
async def replace_item(item_id: int):
    return {"id": item_id, "replaced": True}

@app.patch("/items/{item_id}")
async def update_item(item_id: int):
    return {"id": item_id, "patched": True}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"deleted": item_id}

PUT replaces the full resource. PATCH does a partial update. GET reads. POST creates. DELETE deletes. Standard REST stuff — FastAPI doesn’t invent anything new here, it just makes the decorators dead simple.

Adding metadata for the docs

The decorator accepts a bunch of kwargs that flow straight into the OpenAPI spec:

@app.post(
    "/orders/",
    status_code=201,
    summary="Create a new order",
    description="Creates an order for the authenticated user. Returns 201 with the new order ID.",
    tags=["orders"],
    response_description="The created order",
)
async def create_order():
    return {"id": "ord_123", "status": "pending"}

Each of these shows up in /docs:

• summary — short title, shown collapsed.
• description — full Markdown body, shown when expanded.
• tags — group endpoints into sections in Swagger UI.
• status_code — default response code (we still return the body normally).
• response_description — text next to the response example.

Alternatively, we can put the description in the function docstring and FastAPI picks it up:

@app.post("/orders/", tags=["orders"])
async def create_order():
    """
    Creates an order for the authenticated user.

    Returns 201 with the new order ID.
    """
    return {"id": "ord_123"}

Order matters

Routes match in the order they’re declared. If we have a fixed path and a parameterized path that could overlap, the fixed one must come first.

@app.get("/users/me")
async def current_user():
    return {"id": "self", "name": "current"}

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"id": user_id}

If we swapped these, /users/me would hit get_user with user_id = "me" and FastAPI would 422 because “me” isn’t an int. Always declare more specific routes first.

Deprecating an endpoint

We can mark a route deprecated without removing it. It still works, but Swagger shows it struck-through.

@app.get("/legacy/items/", deprecated=True)
async def old_list():
    return {"warning": "use /items/ instead"}

Multiple methods on one handler

Rare but possible — we can use app.api_route:

@app.api_route("/health", methods=["GET", "HEAD"])
async def health():
    return {"status": "ok"}

Routers (preview)

For real apps we don’t dump every route on app directly. We use APIRouter to split routes across files:

from fastapi import APIRouter

router = APIRouter(prefix="/items", tags=["items"])

@router.get("/")
async def list_items():
    return []

# In main.py
app.include_router(router)

Same decorators, but they live on the router. We’ll dig into routers in a later note.

The interview takeaway

Path operations are just decorators that register a function for an HTTP method + path. The decorator’s kwargs (tags, summary, status_code, deprecated) feed directly into the OpenAPI doc. Watch route order — specific before parameterized.

Path parameters are the dynamic chunks of a URL. In /users/42, the 42 is a path parameter. FastAPI captures them with curly braces in the path string and matches them to function arguments by name.

The basic shape

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id, "type": type(user_id).__name__}

The name in {user_id} must match the function parameter name. The type hint int does two things:

1. FastAPI converts the string from the URL into an int before calling the function.
2. If the conversion fails, FastAPI returns 422 with a clear error — we never see the bad value.

curl http://localhost:8000/users/42
# {"user_id":42,"type":"int"}

curl http://localhost:8000/users/abc
# 422 - {"detail":[{"type":"int_parsing","loc":["path","user_id"],...}]}

Supported types

The usual suspects work out of the box:

• int — /items/5
• float — /prices/19.99
• str — /slug/hello-world (the default; matches any non-slash text)
• bool — /feature/true (accepts “true”, “false”, “1”, “0”, “yes”, “no”)
• UUID — /orders/123e4567-e89b-12d3-a456-426614174000

Validating with Path()

For deeper validation (min/max, regex, etc.) we use the Path() helper:

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(
        ...,
        title="Item ID",
        ge=1,            # greater than or equal to 1
        le=10000,        # less than or equal to 10000
        description="The numeric ID of the item",
    ),
):
    return {"item_id": item_id}

The ... (Ellipsis) means “required, no default.” Available validators:

• ge, gt, le, lt — numeric bounds
• min_length, max_length — string bounds
• pattern — regex constraint

@app.get("/products/{sku}")
async def get_product(sku: str = Path(..., pattern=r"^SKU-\d{6}$")):
    return {"sku": sku}

If someone hits /products/banana they get a 422. Only SKU-123456 style values pass through.

Predefined values with Enum

If we want a path param to only accept a fixed set of strings, use a str Enum. FastAPI shows a dropdown in /docs:

from enum import Enum

class OrderStatus(str, Enum):
    pending = "pending"
    shipped = "shipped"
    delivered = "delivered"

@app.get("/orders/status/{status}")
async def list_by_status(status: OrderStatus):
    if status is OrderStatus.shipped:
        return {"message": "Showing shipped orders"}
    return {"message": f"Showing {status.value} orders"}

Important: the Enum must inherit from str (or int) so OpenAPI knows the underlying primitive. Try any value outside the enum and we get 422.

Path containing slashes

By default str path params stop at the next /. If we genuinely need to match a path like /files/foo/bar/baz.txt, we tell FastAPI with :path:

@app.get("/files/{file_path:path}")
async def read_file(file_path: str):
    return {"file_path": file_path}

Now file_path captures everything after /files/, slashes included. Useful for proxying file paths or wiki-style URLs.

Multiple path params

Just add more {} and matching args:

@app.get("/users/{user_id}/orders/{order_id}")
async def get_user_order(user_id: int, order_id: int):
    return {"user_id": user_id, "order_id": order_id}

Order in the function signature doesn’t matter — FastAPI matches by name.

Common interview gotcha

If asked “what if the path param doesn’t match the type?” — FastAPI returns 422 with a structured error before our handler ever runs. We don’t have to write a try/except. That’s the whole pitch: validation happens at the framework boundary.

Query parameters are the ?key=value&other=thing bits after the URL path. In FastAPI we read them by declaring function arguments that aren’t in the path string. That’s it — no request.query.get() ceremony.

The basic rule

If we declare a function param whose name doesn’t appear in the route path, FastAPI treats it as a query param.

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit, "items": [...]}

Hit /items/?skip=20&limit=50 and FastAPI parses both into ints. Defaults make them optional — call /items/ and we get skip=0, limit=10.

Required vs optional

A query param is optional if it has a default value. It’s required if it doesn’t.

@app.get("/search/")
async def search(q: str):            # required
    return {"q": q}

@app.get("/search/v2/")
async def search_v2(q: str = ""):    # optional, defaults to empty
    return {"q": q}

@app.get("/search/v3/")
async def search_v3(q: str | None = None):  # optional, may be None
    return {"q": q}

str | None = None (or Optional[str] = None) is the pattern when “missing” is genuinely different from “empty string.”

Type coercion

Booleans are forgiving:

@app.get("/items/")
async def list_items(in_stock: bool = False):
    return {"in_stock": in_stock}

?in_stock=true, ?in_stock=1, ?in_stock=yes, ?in_stock=on all parse to True. Their false counterparts go to False. Anything else gets a 422.

Validation with Query()

For length, regex, or numeric constraints, use the Query() helper:

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/products/")
async def search_products(
    q: str = Query(
        ...,                       # ... means required
        min_length=3,
        max_length=50,
        title="Search query",
        description="Free-text search across product names",
    ),
    page: int = Query(1, ge=1, le=1000),
    sort: str = Query("created", pattern="^(created|price|name)$"),
):
    return {"q": q, "page": page, "sort": sort}

• min_length, max_length — string length bounds.
• pattern — regex match.
• ge, gt, le, lt — numeric bounds.
• title, description — show up in /docs.

List query parameters

Some APIs need to accept multiple values for the same key: ?tag=python&tag=fastapi. We type the param as a list:

@app.get("/articles/")
async def list_articles(tag: list[str] = Query(default=[])):
    return {"tags": tag}

A request like /articles/?tag=python&tag=fastapi&tag=backend gives us tag = ["python", "fastapi", "backend"].

Without Query(default=[]), FastAPI would think list[str] is a request body, not a query param. The explicit Query() is what tells it “this is a query param that can repeat.”

Alias for keys with weird characters

OpenAPI / URL conventions often use kebab-case but Python uses snake_case. The alias keeps both happy:

@app.get("/items/")
async def list_items(item_query: str | None = Query(default=None, alias="item-query")):
    return {"item_query": item_query}

The URL sees ?item-query=hello, the function gets item_query="hello".

Deprecation

Mark a query param deprecated and Swagger shows it crossed out:

@app.get("/items/")
async def list_items(q: str | None = Query(default=None, deprecated=True)):
    return {"q": q}

Combining path + query

Path params and query params coexist naturally:

@app.get("/users/{user_id}/orders/")
async def user_orders(
    user_id: int,
    status: str | None = None,
    limit: int = 20,
):
    return {"user_id": user_id, "status": status, "limit": limit}

user_id is in the path string so it’s a path param. status and limit aren’t, so they’re query params. FastAPI figures it out from the route declaration alone.

Interview tip

The “is this a path param or query param?” question comes up a lot. The rule: if the parameter name appears in the URL pattern between {}, it’s a path param. Otherwise (assuming it’s a simple type, not a Pydantic model) it’s a query param.

A request body is the JSON (or form, or file) that a client sends along with a POST/PUT/PATCH. In FastAPI we describe the expected shape with a Pydantic model — that gives us validation, serialization, and OpenAPI schema all in one go.

The basic model

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    in_stock: bool = True

@app.post("/items/")
async def create_item(item: Item):
    return {"created": item, "total_with_tax": item.price * 1.18}

Three things happen automatically:

1. FastAPI reads the request body as JSON.
2. Validates it against Item — wrong types or missing required fields → 422.
3. Hands us a real Item instance with attribute access (item.price, not item["price"]).

Send this:

{
  "name": "Notebook",
  "price": 4.50
}

And we get back:

{
  "created": {"name": "Notebook", "description": null, "price": 4.5, "in_stock": true},
  "total_with_tax": 5.31
}

How FastAPI decides “is this a body param?”

The rule of thumb in simple language:

• Param name in {} → path param.
• Simple type (int, str, bool, etc.) → query param.
• Pydantic model → request body.

Nested models

Pydantic composes nicely. Models can hold other models:

class Address(BaseModel):
    street: str
    city: str
    zip_code: str

class User(BaseModel):
    name: str
    email: str
    addresses: list[Address] = []

@app.post("/users/")
async def create_user(user: User):
    return user

The OpenAPI schema in /docs will show the full nested shape.

Multiple body params

We can declare more than one Pydantic param and FastAPI merges them into a single JSON body, keyed by the param names:

class Item(BaseModel):
    name: str
    price: float

class User(BaseModel):
    username: str

@app.post("/orders/")
async def place_order(item: Item, user: User):
    return {"item": item, "user": user}

Expected body:

{
  "item": {"name": "Pen", "price": 2.5},
  "user": {"username": "manish"}
}

Note the nesting — item and user become top-level keys. This is different from Flask where we’d flatten everything.

Body() for embedding singletons

What if we want a single scalar in the body — not a path or query param? Use Body():

from fastapi import Body

@app.post("/orders/{order_id}/notes/")
async def add_note(order_id: int, note: str = Body(...)):
    return {"order_id": order_id, "note": note}

Without Body(), FastAPI would treat note: str as a query param. With it, note lives in the JSON body. Expected:

{"note": "Customer prefers afternoon delivery"}

Embedding a single Pydantic model

Normally one Pydantic param becomes the entire body. If we want to wrap it under its name (for API consistency, or to add fields later), pass embed=True:

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item = Body(..., embed=True)):
    return {"item_id": item_id, "item": item}

Now the body has to look like:

{
  "item": {"name": "Pen", "price": 2.5}
}

Instead of just the raw item fields. Useful when our API standardizes on {"resource_name": {...}} envelopes.

Field-level validation

For per-field rules, use Field inside the model — same vibe as Query and Path:

from pydantic import BaseModel, Field

class Product(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0, description="Must be positive")
    sku: str = Field(pattern=r"^SKU-\d{6}$")

Pydantic handles all of these before our handler sees the request.

Interview takeaway

Pydantic is the heart of FastAPI’s developer experience. We declare data shapes once as a BaseModel and get parsing, validation, serialization, type hints, and OpenAPI schema in one shot. When asked “how does FastAPI validate request bodies?” — the answer is “it delegates to Pydantic at the framework boundary.”

A Pydantic model is a Python class where the type annotations ARE the schema. We write what we want, Pydantic validates and converts incoming data, and FastAPI uses the same model to generate OpenAPI docs.

In simple language — we describe our data once using normal Python type hints, and we get three things free: validation, JSON serialization, and auto-generated API docs.

Why we care

Without Pydantic, every endpoint becomes a wall of if not isinstance(...) checks. With it, we just declare the shape and FastAPI rejects bad requests before our handler runs.

The flow looks like this:

The basics

We inherit from BaseModel and annotate fields. That’s it.

from pydantic import BaseModel
from datetime import datetime

class User(BaseModel):
    id: int
    email: str
    is_active: bool = True
    created_at: datetime | None = None

Pass it raw data and Pydantic coerces where it safely can (string "42" becomes int 42), or raises ValidationError.

user = User(id="42", email="manish@example.com")
print(user.id)  # 42 (int, not str)

Use it in a FastAPI handler

Just declare the model as a parameter type. FastAPI does the rest — body parsing, validation, docs.

from fastapi import FastAPI

app = FastAPI()

@app.post("/users")
def create_user(user: User):
    # user is already validated here
    return {"created": user.email}

Dumping back to dict / JSON

We constantly need to serialize models — for DB inserts, logs, responses.

user.model_dump()       # -> dict
user.model_dump_json()  # -> JSON string
user.model_dump(exclude={"created_at"})
user.model_dump(exclude_none=True)  # drop None fields

Pydantic v1 vs v2 — the diff that bites everyone

FastAPI now requires v2. If we copy old StackOverflow answers, things break. The big renames:

v2 is also ~17x faster — the core is written in Rust now. So the migration pain is worth it.

ConfigDict — model-level settings

When we need to tweak behavior (allow ORM objects, strip whitespace, forbid extra fields):

from pydantic import BaseModel, ConfigDict

class User(BaseModel):
    model_config = ConfigDict(
        from_attributes=True,   # read from SQLAlchemy objects
        str_strip_whitespace=True,
        extra="forbid",         # reject unknown fields
    )

    id: int
    email: str

The takeaway — Pydantic models are the contract between the outside world and our handlers. We describe the shape once, get validation, docs, and serialization for free.

Type hints get us “this must be a string”. Field() gets us “this must be a string between 3 and 50 characters, matching a slug pattern”. It’s how we encode business rules right next to the field.

In simple language — Field() is the place we attach all the little rules a value must satisfy, plus extras like example values for the docs.

Why bother

We could write a manual validator for every constraint, but that’s noisy. Field() covers 90% of common cases declaratively, and the constraints show up in the OpenAPI schema — Swagger UI displays “min length: 3” automatically.

The shape of Field()

from pydantic import BaseModel, Field
from uuid import uuid4

class CreateUser(BaseModel):
    username: str = Field(min_length=3, max_length=20, pattern=r"^[a-z0-9_]+$")
    age: int = Field(gt=0, lt=150)
    email: str = Field(..., examples=["manish@example.com"])
    bio: str | None = Field(default=None, max_length=500)
    id: str = Field(default_factory=lambda: str(uuid4()))

Two things to spot:

• ... (Ellipsis) means “required, no default” — same as just email: str with no default.
• default_factory is for mutable defaults (lists, dicts, UUIDs). Never use default=[] — that list is shared across instances.

Constraint cheat sheet

A realistic example

Product catalog endpoint. Notice how the constraints document the API for free:

from pydantic import BaseModel, Field
from decimal import Decimal

class CreateProduct(BaseModel):
    sku: str = Field(min_length=8, max_length=12, pattern=r"^[A-Z0-9-]+$")
    name: str = Field(min_length=1, max_length=200)
    price: Decimal = Field(gt=0, le=1_000_000, decimal_places=2)
    stock: int = Field(ge=0, default=0)
    tags: list[str] = Field(default_factory=list, max_length=10)
    description: str | None = Field(default=None, max_length=2000)

@app.post("/products")
def create(product: CreateProduct):
    return {"sku": product.sku}

Send price: -5 and FastAPI returns 422 before our handler runs:

{
  "detail": [{
    "loc": ["body", "price"],
    "msg": "Input should be greater than 0",
    "type": "greater_than"
  }]
}

default vs default_factory — the gotcha

Mutable defaults are evaluated once at class definition. Every instance shares the same list:

# BAD — shared list across all instances
class Cart(BaseModel):
    items: list[str] = Field(default=[])

# GOOD — fresh list every time
class Cart(BaseModel):
    items: list[str] = Field(default_factory=list)

Pydantic actually catches this and raises an error for mutable defaults in most cases, but the rule is — anytime the default needs to be “computed” (timestamp, UUID, empty list), use default_factory.

Extra metadata for docs

Field() also takes description, title, examples, deprecated. All of it flows into Swagger UI:

class CreateUser(BaseModel):
    email: str = Field(
        description="User's primary email, must be unique",
        examples=["manish@example.com"],
    )

The takeaway — type hints describe what type, Field() describes what values are acceptable. Together they encode the entire input contract.

Field() handles simple constraints. For anything fancier — “password must contain a digit AND a special char”, “end_date must be after start_date”, “username must not be a banned word” — we need custom validators.

In simple language — a validator is a method on the model that runs during parsing. We get the value, we either return it (possibly transformed) or raise a ValueError.

Two kinds of validators

@field_validator — single field

The classic case: validating password strength.

from pydantic import BaseModel, field_validator
import re

class SignUp(BaseModel):
    username: str
    password: str

    @field_validator("password")
    @classmethod
    def password_must_be_strong(cls, v: str) -> str:
        if len(v) < 8:
            raise ValueError("password must be at least 8 chars")
        if not re.search(r"\d", v):
            raise ValueError("password must contain a digit")
        if not re.search(r"[!@#$%^&*]", v):
            raise ValueError("password must contain a special character")
        return v

Things to notice:

• @classmethod is required in v2 (it’s gone in v1).
• We return the value — we can transform it (lowercase it, strip it) and the returned value is what gets stored.
• Raise ValueError, not HTTPException. Pydantic catches it and bundles it into the 422 response.

Validate multiple fields at once

Pass several field names to the same validator:

@field_validator("username", "email")
@classmethod
def no_whitespace(cls, v: str) -> str:
    if " " in v:
        raise ValueError("must not contain spaces")
    return v.lower()

mode=“before” vs mode=“after”

By default, validators run after Pydantic has done its type coercion. Sometimes we want to run before — to handle raw input ourselves.

class Event(BaseModel):
    tags: list[str]

    @field_validator("tags", mode="before")
    @classmethod
    def split_string(cls, v):
        # Accept "python,fastapi,backend" as well as a list
        if isinstance(v, str):
            return [t.strip() for t in v.split(",")]
        return v

mode="before" gets raw input. mode="after" (default) gets the already-typed value.

@model_validator — cross-field rules

When the rule needs two or more fields, @field_validator can’t help — each only sees its own value. Use @model_validator instead.

from pydantic import BaseModel, model_validator
from datetime import date

class Booking(BaseModel):
    start_date: date
    end_date: date
    guests: int

    @model_validator(mode="after")
    def check_dates(self) -> "Booking":
        if self.end_date <= self.start_date:
            raise ValueError("end_date must be after start_date")
        if self.guests > 1 and (self.end_date - self.start_date).days < 1:
            raise ValueError("multi-guest bookings need at least 1 night")
        return self

With mode="after", the validator receives self — a fully-built model instance. We return self (or a modified one). With mode="before", we receive the raw dict before any field validation runs.

Validators run in order

Field validators run in field declaration order. Model validators run after all field validators. Knowing this matters when one field depends on another being already validated.

The takeaway — use @field_validator for single-field rules with logic too custom for Field(). Reach for @model_validator only when fields must agree with each other.

Real data isn’t flat. A user has an address. An order has line items. A response wraps a list with pagination metadata. Pydantic lets us nest models inside models — and validates the whole tree in one shot.

In simple language — once we have a model, we can use it as a field type in another model. Pydantic handles parsing recursively, and FastAPI auto-generates nested OpenAPI schemas.

The basic shape

from pydantic import BaseModel

class Address(BaseModel):
    street: str
    city: str
    zip_code: str
    country: str = "IN"

class User(BaseModel):
    id: int
    name: str
    address: Address  # nested model as a field type

Send this JSON to a POST /users:

{
  "id": 1,
  "name": "Manish",
  "address": {
    "street": "MG Road",
    "city": "Bangalore",
    "zip_code": "560001"
  }
}

Pydantic builds the Address instance, attaches it to user.address, and we access it as user.address.city. If the address is missing a field or has a wrong type — 422 before our handler.

Lists of models

For an order with line items, the field type is list[LineItem].

class LineItem(BaseModel):
    product_id: int
    quantity: int
    unit_price: float

class Order(BaseModel):
    id: int
    customer_email: str
    items: list[LineItem]
    total: float

@app.post("/orders")
def create_order(order: Order):
    # order.items is a list of LineItem instances
    computed = sum(i.quantity * i.unit_price for i in order.items)
    return {"items": len(order.items), "computed_total": computed}

Every dict inside the JSON array is validated as a LineItem. If one of them has quantity: "two", we get a 422 pointing at body.items.3.quantity.

Optional and deeply nested

We can mix Optional, list, dict freely:

class Coordinates(BaseModel):
    lat: float
    lon: float

class Address(BaseModel):
    street: str
    city: str
    coords: Coordinates | None = None

class Company(BaseModel):
    name: str
    headquarters: Address
    offices: list[Address] = []
    metadata: dict[str, str] = {}

FastAPI walks the whole tree to generate the OpenAPI schema. Swagger UI shows it as a collapsible nested form.

Recursive models — self-referencing

Comment threads, file trees, org charts — anywhere a model contains itself. In v2 we use a string forward reference:

class Comment(BaseModel):
    id: int
    text: str
    replies: list["Comment"] = []

# Optional in v2 — Pydantic resolves it automatically in most cases.
# If you hit a NameError, call:
Comment.model_rebuild()

Send a tree of arbitrary depth and Pydantic validates every level:

{
  "id": 1,
  "text": "Top level",
  "replies": [
    {"id": 2, "text": "A reply", "replies": [
      {"id": 3, "text": "Nested reply", "replies": []}
    ]}
  ]
}

Serialization handles nesting too

model_dump() recursively converts the whole tree to plain dicts:

order.model_dump()
# {
#   "id": 1,
#   "customer_email": "...",
#   "items": [{"product_id": 7, "quantity": 2, "unit_price": 19.99}, ...]
# }

Why this is a big deal for FastAPI

Each nested model becomes a $ref in the OpenAPI spec. That means:

• Swagger UI shows expandable nested forms.
• Auto-generated client SDKs (TypeScript, Python) get typed nested objects.
• Errors are pinpointed by JSON path — body.address.zip_code not just “bad input”.

The takeaway — model composition is just type-hinting models inside models. Pydantic handles the recursion for parsing AND serialization, and FastAPI propagates the full shape to the API docs.

Depends() is the single feature that makes FastAPI feel different from Flask or Express. We declare what our handler needs — a DB session, a current user, parsed pagination params — and FastAPI builds those things and injects them. We never call them ourselves.

In simple language — instead of doing user = get_current_user(token) at the top of every handler, we put user: User = Depends(get_current_user) in the signature and FastAPI does it for us, every request.

The core idea

A dependency is just a function. Its parameters can be other request data (Header, Query, Body) or other dependencies. Whatever it returns gets passed into our handler.

A first example — pagination params

We want ?skip=0&limit=10 on a bunch of list endpoints. Without DI, we copy-paste the same query params everywhere.

from fastapi import FastAPI, Depends

app = FastAPI()

def pagination(skip: int = 0, limit: int = 10) -> dict:
    return {"skip": skip, "limit": min(limit, 100)}

@app.get("/users")
def list_users(page: dict = Depends(pagination)):
    return {"params": page}

@app.get("/products")
def list_products(page: dict = Depends(pagination)):
    return {"params": page}

Both endpoints now accept skip and limit, the cap on limit is in one place, and Swagger picks them up automatically.

A realistic example — current user from JWT

This is the pattern everyone uses for auth.

from fastapi import Header, HTTPException, Depends

def get_current_user(authorization: str = Header(...)) -> dict:
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token")
    token = authorization.removeprefix("Bearer ")
    user = decode_jwt(token)  # imagine this exists
    if not user:
        raise HTTPException(401, "invalid token")
    return user

@app.get("/me")
def me(user: dict = Depends(get_current_user)):
    return user

@app.post("/posts")
def create_post(payload: PostIn, user: dict = Depends(get_current_user)):
    return {"author": user["id"], "title": payload.title}

The Depends(get_current_user) line on every protected endpoint is literally the auth check. If the dependency raises HTTPException, the handler never runs.

The two ways to write it

These are equivalent — pick whichever your team prefers:

# Style 1: in default value
def handler(user: User = Depends(get_current_user)): ...

# Style 2: with Annotated (FastAPI 0.95+, recommended)
from typing import Annotated
CurrentUser = Annotated[User, Depends(get_current_user)]

def handler(user: CurrentUser): ...

The Annotated form lets us alias common dependencies and reuse them. Much cleaner once the codebase grows.

Per-request caching — the magic

If the same dependency appears multiple times in one request (in the handler AND in another dependency), FastAPI calls it once and reuses the result.

This is huge — we can use the same get_db dependency in three places and still get one connection per request, not three.

To disable caching for a specific dependency, use Depends(get_db, use_cache=False).

Path / router-level dependencies

When a dependency exists only for its side effects (rate limit check, audit log), we don’t want its return value cluttering the handler signature. Attach it at the route or router level:

def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != "secret":
        raise HTTPException(403)

# Single route
@app.get("/admin/stats", dependencies=[Depends(verify_api_key)])
def stats():
    return {...}

# Entire router
from fastapi import APIRouter
router = APIRouter(dependencies=[Depends(verify_api_key)])

Why this beats decorators

Frameworks like Flask solve auth with @login_required decorators. The problem — decorators are opaque, can’t be composed cleanly, can’t be type-hinted, and don’t show up in OpenAPI docs.

Depends() is a regular value in the signature. Type checkers see it, IDE autocompletes it, the result is statically typed, and the params it depends on (Header, Query) appear in Swagger UI automatically. We compose dependencies like we compose any other function.

The takeaway — Depends() flips control. Instead of imperatively pulling data inside the handler, we declaratively list what the handler needs and let FastAPI assemble it. Caching, error propagation, and docs all come along for free.

A dependency doesn’t have to be a function. Anything callable works — including a class. When we pass a class to Depends(), FastAPI calls ClassName(...) for each request and injects the resulting instance.

In simple language — if our “dependency” is just a few parameters bundled together, or if it needs configuration we set at app startup, a class fits better than a function.

Why classes over functions

Functions are great for stateless dependencies (parse pagination, decode a token). But sometimes:

• We want grouped parameters with attribute access (page.skip reads nicer than page["skip"]).
• The dependency needs configuration injected at app startup (DB pool, Redis client, config object).
• We want a clear, mockable interface for tests.

Classes give us all three.

Grouped params — the simplest case

Functions return a dict. Classes return an instance with typed attributes:

from fastapi import Depends, FastAPI

app = FastAPI()

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = min(limit, 100)

@app.get("/users")
def list_users(page: Pagination = Depends(Pagination)):
    return {"skip": page.skip, "limit": page.limit}

FastAPI inspects __init__ exactly like it would inspect a function signature — skip and limit become query params, defaults flow through, Swagger picks them up.

The shortcut

When the dependency IS the class itself, repeating it feels silly:

def list_users(page: Pagination = Depends(Pagination)):  # noisy

FastAPI lets us write it as:

def list_users(page: Pagination = Depends()):  # cleaner

Same behavior, half the typing.

Callable instances — the shared-state pattern

Here’s the killer pattern. We instantiate the class once at startup with config (DB URL, redis client, feature flag), then the instance itself becomes the dependency. Per request, FastAPI calls the instance (via __call__), not the class.

import redis
from fastapi import Depends, HTTPException, Header

class RateLimiter:
    def __init__(self, redis_client, max_per_minute: int):
        self.r = redis_client
        self.max = max_per_minute

    def __call__(self, x_api_key: str = Header(...)) -> str:
        key = f"ratelimit:{x_api_key}"
        count = self.r.incr(key)
        if count == 1:
            self.r.expire(key, 60)
        if count > self.max:
            raise HTTPException(429, "too many requests")
        return x_api_key

# Build once at startup
rate_limit = RateLimiter(redis.Redis(), max_per_minute=60)

@app.get("/search")
def search(api_key: str = Depends(rate_limit)):
    return {"caller": api_key}

Why this rules:

• The Redis client is created once, not per request.
• The max_per_minute is config we control at startup.
• The class is trivial to swap in tests — instantiate with a fake redis and a different limit.

Testability — the real argument

Compare. With a function:

def get_db() -> Session:
    return Session(global_engine)  # hardcoded global

Testing this means monkey-patching the global. Ugly.

With a class:

class DBProvider:
    def __init__(self, engine):
        self.engine = engine
    def __call__(self) -> Session:
        return Session(self.engine)

db_provider = DBProvider(engine=prod_engine)

# In tests
app.dependency_overrides[db_provider] = DBProvider(engine=test_engine)

We swap the entire provider at the DI boundary. Zero monkey-patching, zero globals.

Sub-deps work the same

A class can declare other Depends() in __init__ (or __call__ for callable instances):

class AuditedDB:
    def __init__(
        self,
        db: Session = Depends(get_db),
        user: User = Depends(get_current_user),
    ):
        self.db = db
        self.user = user

    def log_and_query(self, sql: str):
        audit(self.user.id, sql)
        return self.db.execute(sql)

The takeaway — function dependencies for simple stateless work, class dependencies (especially callable instances) when we need startup config, grouped state, or clean test seams. The Depends() API is the same; only the callable changes.

A dependency can declare its own dependencies. Those can declare more. FastAPI walks the whole tree, builds everything in the right order, and threads the results through.

In simple language — Depends() composes. We chain small focused dependencies (read header → decode token → fetch user → check role) and each handler just declares the top of the chain.

The chain

A common auth pipeline. Each layer has one job:

from fastapi import Depends, Header, HTTPException

def get_token(authorization: str = Header(...)) -> str:
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "missing bearer token")
    return authorization.removeprefix("Bearer ")

def get_current_user(token: str = Depends(get_token)) -> User:
    user = decode_and_lookup(token)
    if not user:
        raise HTTPException(401, "invalid token")
    return user

def require_admin(user: User = Depends(get_current_user)) -> User:
    if user.role != "admin":
        raise HTTPException(403, "admin only")
    return user

@app.delete("/users/{user_id}")
def delete_user(user_id: int, admin: User = Depends(require_admin)):
    return {"deleted_by": admin.email}

Our handler asks for require_admin. FastAPI sees it needs get_current_user, which needs get_token, which needs the Authorization header. The whole chain runs in order. Any layer raising stops the request.

The dependency graph

For a single endpoint:

Per request, top to bottom: read header, decode token, fetch user, check role, run handler. Each layer is independently testable.

Request-scoped caching — the multi-edge case

When the dependency graph is a tree, things are simple. When it’s a diamond — the same dep used by multiple branches — caching matters.

Imagine the handler depends on get_current_user AND uses require_admin (which also depends on get_current_user). Without caching we’d call get_current_user twice — meaning two DB lookups for the same user.

FastAPI caches every dependency per request. The first call runs it; every subsequent reference returns the cached value.

@app.get("/admin/dashboard")
def dashboard(
    user: User = Depends(get_current_user),     # uses cached result
    admin: User = Depends(require_admin),       # require_admin also calls get_current_user, cached
):
    # user and admin are literally the same object
    assert user is admin
    return {...}

This is enormous in practice. We stitch together small dependencies fearlessly because shared sub-deps don’t multiply work.

Opting out of the cache

Rarely we want a fresh call every time the dep is referenced — for instance, a “request ID” generator that should produce a new UUID per slot. Set use_cache=False:

def fresh_uuid() -> str:
    return str(uuid4())

@app.get("/trace")
def trace(
    a: str = Depends(fresh_uuid, use_cache=False),
    b: str = Depends(fresh_uuid, use_cache=False),
):
    return {"a": a, "b": b}  # different UUIDs

Default is use_cache=True and that’s almost always what we want.

The scope of “request”

The cache lives for one HTTP request — start to response. The next request starts with an empty cache. There’s no cross-request caching at the DI level (use Redis or a module-level cache for that).

This per-request scope is also why DB sessions work cleanly with Depends — same session for the whole request, fresh session for the next one.

Composition wins

Sub-dependencies turn DI into a programming model. Auth check, feature flag check, tenant scoping, current org lookup — each is a five-line function. Endpoints declare only the top of the chain. New endpoint? Add Depends(require_admin) and we’re done.

The takeaway — Depends() composes recursively. FastAPI resolves the whole graph per request and caches each node. We get small, testable, composable middleware-without-middleware.

A regular dependency returns a value. A yield dependency yields a value, runs the handler, then resumes after the yield to clean up. Think of it like a context manager (with block) wrapped around the request.

In simple language — anywhere we’d write with session() as s: in a script, we use a yield dependency in FastAPI. It guarantees teardown — close the DB session, release the lock, log the timing — no matter what happens in the handler.

The shape

from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db          # <-- value injected into handler
    finally:
        db.close()        # <-- runs AFTER the response is built

The yield splits the function in two:

• Code before yield runs on request start (setup).
• The value after yield is injected.
• Code after yield runs on request end (teardown), even if the handler raised.

Lifecycle in pictures

The canonical use case — SQLAlchemy session

This is in every FastAPI codebase that talks to a DB:

from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(404)
    return user

Every request gets its own session. The finally block closes it — whether the handler returned cleanly, raised HTTPException, or blew up with an unexpected error.

Important: always put cleanup in finally, not after the bare yield. If we skip finally, an exception in the handler skips the cleanup.

Exception handling INSIDE the yield

We can catch exceptions raised in the handler and react before re-raising (or instead of re-raising). The pattern — wrap the yield itself.

def db_session_with_rollback():
    db = SessionLocal()
    try:
        yield db
        db.commit()                  # success path
    except Exception:
        db.rollback()                # something went wrong in the handler
        raise                        # let FastAPI's error handlers see it
    finally:
        db.close()

In simple language — yield propagates handler exceptions back into our dependency. We can catch them, do something (rollback, log), and either swallow or re-raise. If we swallow it, FastAPI thinks the handler succeeded — usually not what we want, so almost always raise.

Stacking yield dependencies

Multiple yield deps work like nested context managers. Setup happens outermost-in, teardown happens innermost-out.

def get_logger(request: Request):
    log = make_logger(request_id=request.headers.get("x-request-id"))
    log.info("request start")
    try:
        yield log
    finally:
        log.info("request end")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/orders")
def create_order(
    order: OrderIn,
    db: Session = Depends(get_db),
    log = Depends(get_logger),
):
    log.info("creating order")
    return repository.save(db, order)

Yield + raise after teardown? Not allowed

We can’t raise HTTPException from the teardown side (after yield). By that point the response is already constructed and being sent. If we need to abort, do it in the handler or in the setup (before yield).

Async yield works the same

For async code, use async def:

async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

Or manual:

async def get_redis():
    r = await aioredis.create_redis_pool(...)
    try:
        yield r
    finally:
        r.close()
        await r.wait_closed()

Combining with Depends caching

Same per-request caching applies. If three places in the request use Depends(get_db), the session is opened once and torn down once at the end. We don’t get three sessions just because three places asked for one.

The takeaway — yield dependencies are FastAPI’s answer to “I need setup AND teardown around the request”. Always put cleanup in finally. Catch exceptions in the dep if we need to react (rollback). Stack them freely — teardown order is LIFO, exactly like nested context managers.

FastAPI lets us write route handlers as either def or async def. Both work, both return JSON, but under the hood they run on completely different machinery. Picking the wrong one tanks throughput.

In simple language: async def runs on the event loop (one thread, juggles many requests). Plain def runs on a thread pool (FastAPI offloads it so the event loop stays free). So FastAPI is smart enough to not block — but only if we don’t lie to it.

The rule

• Use async def when our code uses await (async DB driver, httpx.AsyncClient, asyncio.sleep).
• Use plain def when our code is sync (requests, psycopg2, time.sleep, CPU work).
• Never mix: don’t call blocking code inside async def without offloading it.

The killer pitfall

A blocking call inside async def freezes the entire event loop. Every other request in flight stalls. This is the single most common FastAPI perf bug.

import time
import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# BAD — blocks the event loop, every other request waits
@app.get("/bad")
async def bad():
    time.sleep(2)              # sync sleep inside async
    return {"status": "done"}

# GOOD — async sleep yields back to the loop
@app.get("/good")
async def good():
    await asyncio.sleep(2)
    return {"status": "done"}

# ALSO GOOD — sync function, FastAPI runs it on a thread
@app.get("/sync-ok")
def sync_ok():
    time.sleep(2)
    return {"status": "done"}

Real example: calling an external API

import httpx
import requests
from fastapi import FastAPI

app = FastAPI()

# async route + async client = correct
@app.get("/weather/{city}")
async def weather(city: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(f"https://api.weather.com/{city}")
    return r.json()

# sync route + sync client = also correct, just less scalable
@app.get("/weather-sync/{city}")
def weather_sync(city: str):
    r = requests.get(f"https://api.weather.com/{city}")
    return r.json()

Offloading CPU work

If we have a CPU-heavy function and want to call it from an async route, use run_in_threadpool (or a process pool for true parallelism, since the GIL still bites).

from fastapi.concurrency import run_in_threadpool

def heavy_pdf_parse(data: bytes) -> str:
    # expensive, blocking, CPU-bound
    return parse(data)

@app.post("/upload")
async def upload(file: bytes):
    text = await run_in_threadpool(heavy_pdf_parse, file)
    return {"text": text}

Interview cheat sheet

• “Why is my FastAPI app slow?” → Look for blocking calls inside async def.
• “What’s the difference?” → Event loop vs thread pool, both non-blocking from FastAPI’s perspective.
• “Should I make everything async?” → Only if the libraries we use are async. A sync DB call in an async route is worse than a fully sync route.

When we return a dict or a model from a route, FastAPI serializes it as-is. But that’s risky — what if we accidentally return the user’s hashed_password? That’s where response_model comes in. Think of it like a strainer between our function’s return value and the wire.

What response_model does

It re-validates and filters our return value against a Pydantic schema. Fields not declared in the model get dropped. Docs auto-update too.

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr

class UserIn(BaseModel):
    email: EmailStr
    password: str   # incoming

class UserOut(BaseModel):
    email: EmailStr
    # no password field — it cannot leak

app = FastAPI()

@app.post("/users", response_model=UserOut)
def create_user(user: UserIn):
    save_to_db(user)
    return user   # password is silently stripped

status_code

By default, 200 OK. For POST that creates a resource, we want 201 Created. For DELETE with no body, 204 No Content.

from fastapi import FastAPI, status

@app.post("/items", response_model=Item, status_code=status.HTTP_201_CREATED)
def create_item(item: ItemIn):
    return save(item)

@app.delete("/items/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_item(id: int):
    remove(id)
    # don't return anything for 204

Use the status module — way more readable than raw integers.

response_model_exclude_unset

In simple language: only send back fields the user actually set. Useful for PATCH endpoints and large optional models.

class Profile(BaseModel):
    name: str | None = None
    bio: str | None = None
    age: int | None = None

@app.get("/me", response_model=Profile, response_model_exclude_unset=True)
def me():
    return {"name": "Manish"}   # response is just {"name": "Manish"}, no nulls

There’s also response_model_exclude_none (drops fields that are explicitly None) and response_model_exclude={"field"} to drop specific fields.

Multiple response models (union)

Sometimes one route returns different shapes — say, a public profile or a private one based on auth.

from typing import Union

class PublicUser(BaseModel):
    username: str

class PrivateUser(PublicUser):
    email: EmailStr
    last_login: str

@app.get("/users/{id}", response_model=Union[PrivateUser, PublicUser])
def get_user(id: int, is_owner: bool = False):
    user = db.get(id)
    return user if is_owner else {"username": user.username}

For different status codes returning different shapes (like an error model on 404), use the responses parameter:

@app.get(
    "/items/{id}",
    response_model=Item,
    responses={404: {"model": ErrorMessage}},
)
def get_item(id: int):
    ...

Interview cheat sheet

• response_model filters output and updates OpenAPI docs.
• Use it to prevent leaking sensitive fields like hashed_password.
• 201 for create, 204 for delete-no-body, 200 for everything else.
• response_model_exclude_unset=True for clean PATCH responses.
• responses={...} documents alternate status codes / models.

OAuth2 sounds scary. In simple language: it’s just a standard way to say “send me your username and password to a specific URL, and I’ll give you a token. Then send that token on every future request.” FastAPI bakes this in via OAuth2PasswordBearer.

It’s the default auth pattern in FastAPI tutorials and most real apps. SwaggerUI even gets a working “Authorize” button for free.

The flow

  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-tag-beginner);">← access_token (JWT)</div>
  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Server</div>

  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-accent);">GET /users/me<br/>Authorization: Bearer &lt;token&gt;</div>
  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Server</div>

  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-tag-beginner);">← user data</div>
  <div style="border: 1px solid var(--color-border); padding: 8px; border-radius: 4px; text-align: center;">Server</div>
</div>

from datetime import datetime, timedelta, timezone
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt, JWTError
from passlib.context import CryptContext
from pydantic import BaseModel

SECRET_KEY = "swap-this-with-a-real-secret"
ALGORITHM = "HS256"
TOKEN_EXPIRE_MIN = 30

app = FastAPI()
pwd_ctx = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# pretend DB
fake_db = {
    "manish": {
        "username": "manish",
        "hashed_password": pwd_ctx.hash("secret"),
    }
}

class Token(BaseModel):
    access_token: str
    token_type: str

def authenticate(username: str, password: str):
    user = fake_db.get(username)
    if not user or not pwd_ctx.verify(password, user["hashed_password"]):
        return None
    return user

def make_token(data: dict) -> str:
    payload = data.copy()
    payload["exp"] = datetime.now(timezone.utc) + timedelta(minutes=TOKEN_EXPIRE_MIN)
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

@app.post("/token", response_model=Token)
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "bad credentials")
    return {"access_token": make_token({"sub": user["username"]}), "token_type": "bearer"}

def get_current_user(token: str = Depends(oauth2_scheme)):
    cred_err = HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid token")
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username = payload.get("sub")
    except JWTError:
        raise cred_err
    user = fake_db.get(username)
    if not user:
        raise cred_err
    return user

@app.get("/users/me")
def me(user = Depends(get_current_user)):
    return {"username": user["username"]}

# get a token
curl -X POST http://localhost:8000/token \
  -d "username=manish&password=secret"

# use it
curl http://localhost:8000/users/me \
  -H "Authorization: Bearer eyJhbGc..."

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"read:items": "Read items", "admin": "Full access"},
)

@app.get("/items")
def list_items(user = Security(get_current_user, scopes=["read:items"])):
    ...

A JWT (JSON Web Token) is a string that contains a payload of claims, signed by the server. In simple language: it’s a tamper-proof note the server hands the client. The client shows it back on every request, and the server can verify “yes, I signed this” without hitting any DB.

That stateless property is the whole appeal. Sessions need a DB lookup per request. JWTs don’t.

Anatomy

Three base64url-encoded parts joined by dots: header.payload.signature.

The payload is not encrypted. Anyone with the token can decode it. Never put secrets in there — only put claims like user id, role, expiry.

Encoding and decoding

Two popular libraries: python-jose (used in FastAPI docs) and PyJWT. They have nearly identical APIs.

pip install "python-jose[cryptography]"

from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError, ExpiredSignatureError

SECRET = "use-a-long-random-string-from-env"
ALGO = "HS256"

def encode_token(user_id: str, minutes: int = 30) -> str:
    payload = {
        "sub": user_id,
        "exp": datetime.now(timezone.utc) + timedelta(minutes=minutes),
        "iat": datetime.now(timezone.utc),
        "type": "access",
    }
    return jwt.encode(payload, SECRET, algorithm=ALGO)

def decode_token(token: str) -> dict:
    return jwt.decode(token, SECRET, algorithms=[ALGO])

jwt.decode checks the signature, the expiry (exp), and raises ExpiredSignatureError or generic JWTError if anything’s off. We don’t have to check expiry manually.

The get_current_user dependency

This is the bridge between “raw token in header” and “User object in route”. Most FastAPI apps have exactly this:

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_current_user(token: str = Depends(oauth2_scheme)):
    creds_exc = HTTPException(
        status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = decode_token(token)
        user_id = payload.get("sub")
        if user_id is None:
            raise creds_exc
    except ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except JWTError:
        raise creds_exc
    user = db.get_user(user_id)
    if not user:
        raise creds_exc
    return user

@app.get("/me")
def me(user = Depends(get_current_user)):
    return user

Every protected route just adds user = Depends(get_current_user). Composable, testable, swappable.

Refresh tokens — why two tokens?

Access tokens should be short-lived (15-30 min) so a leak isn’t catastrophic. But making users log in every 30 minutes is awful UX. Enter refresh tokens: long-lived (7-30 days), used only to get new access tokens.

The pattern:

@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(401)
    return {
        "access_token": encode_token(user.id, minutes=15),
        "refresh_token": encode_refresh(user.id, days=7),
        "token_type": "bearer",
    }

def encode_refresh(user_id: str, days: int) -> str:
    payload = {
        "sub": user_id,
        "exp": datetime.now(timezone.utc) + timedelta(days=days),
        "type": "refresh",
    }
    return jwt.encode(payload, SECRET, algorithm=ALGO)

class RefreshIn(BaseModel):
    refresh_token: str

@app.post("/refresh")
def refresh(body: RefreshIn):
    try:
        payload = decode_token(body.refresh_token)
        if payload.get("type") != "refresh":
            raise HTTPException(401, "wrong token type")
    except JWTError:
        raise HTTPException(401, "invalid refresh token")
    return {"access_token": encode_token(payload["sub"], minutes=15), "token_type": "bearer"}

Two big rules:

1. Type-tag the tokens. A refresh token must not be accepted as an access token (and vice versa). Check payload["type"].
2. Store refresh tokens server-side if we need revocation. Pure stateless JWTs can’t be revoked — that’s the tradeoff. For “log out everywhere”, keep a jti (token id) in the DB and check it.

Common pitfalls

• exp in seconds, not milliseconds. JWT spec uses Unix seconds. python-jose accepts a datetime and handles it; raw dicts don’t.
• Don’t trust the alg field. Always pass algorithms=[ALGO] to decode. There’s a famous “alg: none” attack otherwise.
• Secret rotation. Plan for it. Accept two secrets during a rotation window.
• Don’t put PII in the payload. It’s readable by anyone holding the token.

Interview cheat sheet

• JWT = signed (not encrypted) JSON. Three parts: header.payload.signature.
• Server can verify without DB lookup → stateless auth.
• Access token short, refresh token long. Type-tag them.
• Revocation needs a server-side allowlist/blocklist — pure JWTs can’t be revoked.
• Always pin the algorithm on decode.
• Use python-jose or PyJWT. Both fine. jose if we want JWE/JWS later.

Middleware is code that wraps every request and response. Think of it like a security guard at a door — every visitor passes through, the guard can inspect, modify, or block them, then lets them through to the actual room. Logging, timing, auth headers, request IDs — all classic middleware jobs.

Custom middleware with @app.middleware("http")

The decorator gives us the simplest API. We receive the request, call the next handler with await call_next(request), get the response, and return it.

import time
import uuid
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start = time.perf_counter()
    request_id = str(uuid.uuid4())
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    response.headers["X-Request-ID"] = request_id
    return response

Middlewares run in reverse order of registration on the way back out. Last added = outermost.

CORS — the browser security thing

Browsers block JavaScript from https://app.com calling https://api.com unless api.com says it’s OK. That’s the same-origin policy. CORS is how the server says “yes, this origin is allowed”.

Without CORS configured, our React app on localhost:3000 cannot call our API on localhost:8000. Browser blocks it before the request even gets a response payload.

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.pman47.cc", "http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Notes:

• Never use allow_origins=["*"] with allow_credentials=True. Browsers reject that combo, and it’s a security footgun.
• allow_methods=["*"] is fine for development. In prod, list them explicitly: ["GET", "POST", "PUT", "DELETE"].
• CORS only matters for browsers. curl and server-to-server calls completely ignore CORS.

Other common built-in middlewares

from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["pman47.cc", "*.pman47.cc"])
app.add_middleware(HTTPSRedirectMiddleware)

Middleware vs Dependencies — when to use which

This trips a lot of people up.

In simple language: middleware is for things every request needs (logging, CORS). Dependencies are for things specific routes need (this endpoint requires a logged-in user).

For auth specifically, dependencies are usually the better call — they show up in OpenAPI docs, work with the /docs Authorize button, and let us mix protected and public routes cleanly.

Order matters

app.add_middleware(CORSMiddleware, ...)         # added second → outer
app.add_middleware(GZipMiddleware, ...)         # added first → inner

CORS should usually be outermost so it handles preflight OPTIONS requests before anything else. If our auth middleware runs before CORS, browser preflights get a 401 and the real request never fires.

Interview cheat sheet

• Middleware = code wrapping every request. Use @app.middleware("http") or app.add_middleware(...).
• CORS = browser thing. Configure CORSMiddleware with explicit origins for prod.
• Middleware vs dependencies: middleware for cross-cutting concerns, dependencies for per-route logic.
• Order matters — last added = outermost. CORS should be outer.
• Curl and server-side clients don’t care about CORS.

Sometimes we want to do work that doesn’t need to block the response. Sending a welcome email after signup is the textbook case — the user shouldn’t wait 2 seconds staring at a spinner just because our SMTP server is slow.

FastAPI ships BackgroundTasks for exactly this. In simple language: “I’ll send the response now, but please run this function right after.”

The pattern

We declare a BackgroundTasks parameter, attach functions to it, and return. FastAPI sends the response, then runs the queued functions.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def send_welcome_email(email: str, name: str):
    # slow, blocking, that's fine — runs after response
    smtp.send(to=email, subject="Welcome!", body=f"Hi {name}")

@app.post("/signup")
def signup(email: str, name: str, tasks: BackgroundTasks):
    user = create_user(email, name)
    tasks.add_task(send_welcome_email, email, name)
    return {"user_id": user.id}   # response goes out NOW

The user sees their response in 50ms. The email sends in the background.

t=0ms   POST /signup arrives
t=20ms  create_user() — DB write
t=22ms  add_task(send_email) — just queues, doesn't run
t=25ms  return response ─► client gets 200 OK
t=25ms  send_welcome_email() starts running
t=2025  email finally sent (client doesn't care)

Sync or async tasks both work

async def push_to_webhook(url: str, payload: dict):
    async with httpx.AsyncClient() as c:
        await c.post(url, json=payload)

@app.post("/event")
def event(tasks: BackgroundTasks, payload: dict):
    save(payload)
    tasks.add_task(push_to_webhook, "https://hooks.slack.com/...", payload)
    return {"ok": True}

Sync tasks run in the thread pool. Async tasks run on the event loop. Same as routes.

Multiple tasks

We can queue several. They run in order.

@app.post("/order")
def place_order(order: Order, tasks: BackgroundTasks):
    saved = save(order)
    tasks.add_task(send_confirmation_email, order.email)
    tasks.add_task(notify_warehouse, saved.id)
    tasks.add_task(update_analytics, "order_placed", saved.id)
    return saved

The big limitations

BackgroundTasks is a great fit for cheap fire-and-forget work, but it has real limits.

1. Runs in the same process. If our server crashes between sending the response and finishing the task, the task is lost. No retries.
2. No persistence. Restart the server, in-flight tasks die.
3. No scheduling. Can’t say “run this in 1 hour”.
4. Uses the same worker. Heavy CPU tasks will eat into request-handling capacity.
5. No visibility. No dashboard, no logs of past tasks, no way to retry a failed task.

When to reach for Celery (or similar)

Use Celery / RQ / Dramatiq / arq when we need:

• Reliability — tasks survive crashes (backed by Redis/RabbitMQ).
• Retries — failed task automatically retries with backoff.
• Scheduling — “send this email tomorrow morning”.
• Heavy work — image processing, ML inference, big PDFs. Don’t bog down API workers.
• Visibility — Flower dashboard, retry counts, success/failure metrics.
• Scale — separate worker pool we can grow independently.

Mental model: BackgroundTasks for “after the response, quickly”. A task queue for “anything important”.

Quick Celery sketch

# tasks.py
from celery import Celery

celery = Celery("app", broker="redis://localhost:6379/0")

@celery.task(bind=True, max_retries=3)
def send_welcome_email(self, email: str, name: str):
    try:
        smtp.send(to=email, subject="Welcome", body=f"Hi {name}")
    except SMTPException as e:
        raise self.retry(exc=e, countdown=60)

# main.py
from tasks import send_welcome_email

@app.post("/signup")
def signup(email: str, name: str):
    user = create_user(email, name)
    send_welcome_email.delay(email, name)   # pushes to Redis, worker picks up
    return {"user_id": user.id}

Run a separate worker process: celery -A tasks worker --loglevel=info. Now restart the API all we want — pending tasks survive in Redis.

arq — the asyncio-native alternative

If our whole stack is async, arq feels more natural than Celery. Same idea, Redis-backed, but built on asyncio from day one.

Interview cheat sheet

• BackgroundTasks runs after the response, in the same process.
• Perfect for fast fire-and-forget work (emails, webhooks, cache invalidation).
• Loses tasks on crash. No retries. No persistence.
• For anything important, use Celery / RQ / arq with a proper broker (Redis/RabbitMQ).
• Rule of thumb: if losing the task would be a bug, use a real queue.

WebSockets are persistent two-way connections between client and server. Unlike HTTP (request → response → done), a WebSocket stays open and both sides can send messages whenever. Think of it like a phone call vs sending letters — once the line is open, anyone can talk.

FastAPI (well, Starlette under the hood) makes WebSockets feel as natural as regular routes.

The basic route

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws/echo")
async def echo(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            msg = await ws.receive_text()
            await ws.send_text(f"echo: {msg}")
    except WebSocketDisconnect:
        print("client disconnected")

Three things to notice:

1. @app.websocket(...) not @app.get(...).
2. Must call await ws.accept() to complete the handshake.
3. Always wrap the loop in try/except WebSocketDisconnect — otherwise we log noisy tracebacks.

Connection lifecycle

  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-tag-beginner);">← 101 Switching Protocols<br/>(ws.accept)</div>
  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Server</div>

  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-accent);">⇄ send / receive loop ⇄<br/>(text, json, bytes)</div>
  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Server</div>

  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Client</div>
  <div style="text-align: center; color: var(--color-tag-advanced);">close frame<br/>(WebSocketDisconnect)</div>
  <div style="border: 1px solid var(--color-border); padding: 6px; border-radius: 4px; text-align: center;">Server</div>
</div>

await ws.send_text("hello")
await ws.send_json({"event": "tick", "value": 42})
await ws.send_bytes(b"\x00\x01\x02")

text = await ws.receive_text()
data = await ws.receive_json()    # parses for us
buf = await ws.receive_bytes()

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        # iterate over a copy — disconnects mutate the list
        for conn in list(self.active):
            try:
                await conn.send_text(message)
            except Exception:
                self.disconnect(conn)

manager = ConnectionManager()
app = FastAPI()

@app.websocket("/ws/chat/{username}")
async def chat(ws: WebSocket, username: str):
    await manager.connect(ws)
    await manager.broadcast(f"-> {username} joined")
    try:
        while True:
            msg = await ws.receive_text()
            await manager.broadcast(f"{username}: {msg}")
    except WebSocketDisconnect:
        manager.disconnect(ws)
        await manager.broadcast(f"<- {username} left")

# sketch — each worker subscribes to a Redis channel
async def listen_redis():
    pubsub = redis.pubsub()
    await pubsub.subscribe("chat")
    async for msg in pubsub.listen():
        if msg["type"] == "message":
            await manager.broadcast(msg["data"].decode())

from fastapi import Query, WebSocket, status

@app.websocket("/ws")
async def secured(ws: WebSocket, token: str = Query(...)):
    user = decode_jwt(token)
    if not user:
        await ws.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await ws.accept()
    ...

FastAPI testing is one of the friendliest stories in Python web frameworks. We don’t spin up a real server — we use an in-process client that calls our app directly. Fast, deterministic, no port conflicts.

TestClient — the workhorse

TestClient is a thin wrapper around httpx that talks to our app via ASGI in-process. We write tests as if we were making real HTTP calls.

pip install pytest httpx

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/health")
def health():
    return {"status": "ok"}

@app.post("/items")
def create_item(item: dict):
    return {"id": 1, **item}

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_health():
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json() == {"status": "ok"}

def test_create_item():
    r = client.post("/items", json={"name": "Book", "price": 10})
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == 1
    assert body["name"] == "Book"