FastAPI Interview Questions and Answers

Answer

FastAPI is a modern Python web framework for building APIs, released in 2018 by Sebastián Ramírez. It solves three problems that older Python frameworks (Flask, Django REST) handled poorly: (1) request validation requires manual code or external libraries, (2) async support is bolted-on rather than native, and (3) API documentation has to be written separately from the code. FastAPI handles all three automatically: Python type hints become Pydantic validation at runtime, every endpoint is async-capable, and OpenAPI/Swagger docs are generated from the function signature.

The framework is built on Starlette (ASGI) and Pydantic, which means it gets concurrency from the former and type safety from the latter. Structurally it is a thin layer: the `FastAPI` class subclasses Starlette's `Starlette`, `APIRouter` subclasses Starlette's `Router`, and `Request`, `Response`, `BackgroundTasks` and `WebSocket` are Starlette objects re-exported under the `fastapi` namespace. What FastAPI genuinely adds is the dependency-injection graph, the signature-to-JSON-Schema compiler, and `response_model` serialisation.

That split matters in interviews: questions about custom ASGI middleware or streaming internals are really Starlette questions. Expect the follow-up about what FastAPI does not give you: no ORM, no migrations, no admin panel, no session auth, no caching or rate limiting. Teams moving off Django REST Framework routinely underestimate that and end up assembling SQLAlchemy, Alembic, Celery and Redis by hand. Version context: release 0.100 (2023) moved the validation core to Pydantic v2, and 0.111 shipped the `fastapi` CLI, so `fastapi dev main.py` is now the documented dev command even though `uvicorn main:app --reload` still works.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(title="Orders API", version="1.0.0")

class Order(BaseModel):
    sku: str
    qty: int

@app.post("/orders", status_code=201)
async def create_order(order: Order):
    # validated, typed, documented, all from the signature
    return {"sku": order.sku, "qty": order.qty}

# fastapi dev main.py   (0.111+, reload enabled)
# uvicorn main:app --reload   (still valid)

Answer

Path parameters are declared by including them in the route decorator and as function arguments with type annotations. Query parameters are any function arguments not in the path. FastAPI parses URL components and converts them to the annotated types, invalid input returns a 422 automatically.

A path parameter is always required because it is part of the URL; a query parameter is optional exactly when you give it a default, so `q: str` is required and `q: str | None = None` is not. Constraints go in `Path()` and `Query()`: `ge`, `le`, `gt`, `lt` for numbers and `min_length`, `max_length`, `pattern` for strings (`pattern` replaced `regex`, which was removed in Pydantic v2 era releases). Details interviewers probe: route matching is first-match-wins in registration order, so `/users/me` must be declared before `/users/{user_id}` or `me` gets parsed as an int and 422s; a value containing slashes needs the Starlette path converter `/{file_path:path}`; repeated query keys bind to a list with `list[str] = Query(default=[])`, and without the list annotation only the last value survives; booleans accept `true`, `True`, `1`, `on` and `yes` case-insensitively.

The 422 body is a JSON object with a `detail` array where each entry carries `type`, `loc` (for example `["path", "item_id"]`), `msg` and `input`, which is what frontend teams parse to attach field-level errors. Enum path parameters (`class Status(str, Enum)`) both validate the value and render a dropdown in Swagger UI.

from enum import Enum
from typing import Annotated
from fastapi import FastAPI, Path, Query

app = FastAPI()

class Status(str, Enum):
    open = "open"
    closed = "closed"

# declare BEFORE /items/{item_id}, otherwise "latest" fails int parsing
@app.get("/items/latest")
async def latest_item():
    return {"item_id": 1}

@app.get("/items/{item_id}")
async def read_item(
    item_id: Annotated[int, Path(ge=1)],
    q: Annotated[str | None, Query(max_length=50)] = None,
    tag: Annotated[list[str], Query()] = [],
    status: Status = Status.open,
):
    return {"item_id": item_id, "q": q, "tag": tag, "status": status}

# GET /items/42?q=hello&tag=a&tag=b -> tag == ["a", "b"]
# GET /items/abc                    -> 422, detail[0].loc == ["path", "item_id"]
# GET /items/0                      -> 422, greater_than_equal

Answer

Pydantic is a Python library for data validation using type hints, originally built by the same author as FastAPI. FastAPI uses Pydantic to validate request bodies, query parameters, and response models. When you declare a function parameter as a Pydantic model, FastAPI: parses the incoming JSON, validates each field against the model's type annotations, returns a structured 422 error if validation fails, and passes a fully-typed model instance to your function.

Pydantic v2 (2023) rewrote the core in Rust as `pydantic-core`, making it 5-50x faster than v1. This is the layer that lets FastAPI feel both magical and safe. Mechanically, each model compiles once at import time into a validator schema, so the per-request cost is Rust-side parsing rather than Python attribute checks; that is why moving validation from hand-written `if` blocks into a model usually gets faster, not slower.

Things that trip candidates up: the default mode is lax, so the JSON string `"30"` coerces into `age: int` and only `model_config = ConfigDict(strict=True)` (or `Field(strict=True)`) stops it; unknown keys are ignored silently unless you set `extra="forbid"`, which is the fix for mass-assignment style bugs; and `EmailStr` raises an import error at startup unless `pydantic[email]` (the `email-validator` package) is installed, a classic Docker build surprise. Interviewers commonly follow up with the difference between `model_validate` (validates) and `model_construct` (skips validation, useful only for trusted internal data), and with why request models and database models should stay separate classes.

from pydantic import BaseModel, ConfigDict, EmailStr, Field
from fastapi import FastAPI

app = FastAPI()

class UserCreate(BaseModel):
    model_config = ConfigDict(extra="forbid")  # reject unknown keys

    email: EmailStr
    age: int = Field(ge=13, le=120)
    full_name: str = Field(min_length=1, max_length=80)

@app.post("/users")
async def create_user(user: UserCreate):
    return {"created": user.model_dump()}

# {"email": "a@b.co", "age": "30", "full_name": "A"}  -> ok, "30" coerced to 30
# {"email": "a@b.co", "age": 30, "is_admin": true}    -> 422 extra_forbidden

Answer

Use the `response_model` parameter on the route decorator. FastAPI will (1) validate your return value against the model, (2) coerce and strip extra fields, (3) generate the response schema in the OpenAPI docs. This is the main defence against leaking internal fields (password hashes, internal IDs, tenant keys) from API responses, because filtering happens even when the handler returns a raw ORM object or dict.

Since FastAPI 0.89 you can use the return type annotation instead (`async def get_user(...) -> UserPublic:`) and the decorator argument is only needed when the two must differ, for example returning a `RedirectResponse` from a route that documents a model. Behaviour worth knowing: filtering is one-way, extra keys are dropped rather than rejected, but a missing required field raises a `ResponseValidationError` and returns 500, which is the intended signal that your handler is broken rather than the client. Reading ORM instances requires `model_config = ConfigDict(from_attributes=True)`, and `response_model_exclude_unset=True` omits fields the caller never set, which is how you keep PATCH responses small. `response_model_by_alias` controls whether camelCase aliases are emitted. Two production notes: serialising a large list through `response_model` costs real CPU, so for 10k-row exports people bypass it with an `ORJSONResponse`; and generics like `Page[UserPublic]` work fine and give the docs a properly named schema, which is nicer than declaring `dict` and losing the contract.

from pydantic import BaseModel, ConfigDict
from fastapi import FastAPI

app = FastAPI()

class UserPublic(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    nickname: str | None = None
    # password_hash is NOT in the public model

# modern form: the return annotation IS the response_model
@app.get("/users/{user_id}")
async def get_user(user_id: int) -> UserPublic:
    return await db.fetch_user(user_id)  # ORM row; password_hash stripped

@app.patch("/users/{user_id}", response_model=UserPublic, response_model_exclude_unset=True)
async def patch_user(user_id: int) -> UserPublic:
    return await db.fetch_user(user_id)

Answer

Use `File()` or `UploadFile` as a parameter type. `UploadFile` is preferred for large files because it streams to disk instead of loading into memory. FastAPI parses `multipart/form-data` automatically. The mechanism: Starlette wraps the upload in a `SpooledTemporaryFile` that stays in RAM up to roughly 1 MB and then rolls over to a real temp file on disk, so `bytes = File()` loads the whole body into memory while `UploadFile` does not. `UploadFile` exposes `filename`, `content_type`, `size`, and async `read`, `write`, `seek` and `close` methods that dispatch to a threadpool, which is why you await them inside `async def`.

Two hard requirements interviewers check: the `python-multipart` package must be installed or the app raises a `RuntimeError` about form data at startup, and you cannot mix a JSON body and file fields in the same request, everything becomes form fields once the request is multipart. Production gotchas: `filename` is attacker-controlled, so never join it into a path without `os.path.basename` plus a generated UUID; `content_type` is client-supplied and must be re-checked from the magic bytes if it matters; and there is no built-in size cap, so enforce one either at the reverse proxy (`client_max_body_size` in nginx) or by counting bytes as you stream chunks and raising 413. Multiple files come through as `list[UploadFile]`, and for anything larger than a few hundred MB the usual answer is a presigned S3 upload so the file never transits the API at all.

import uuid
from fastapi import FastAPI, Form, HTTPException, UploadFile

app = FastAPI()
MAX_BYTES = 10 * 1024 * 1024

@app.post("/upload")
async def upload(file: UploadFile, note: str = Form("")):
    dest = f"/data/{uuid.uuid4()}"
    written = 0
    with open(dest, "wb") as out:
        while chunk := await file.read(1024 * 1024):
            written += len(chunk)
            if written > MAX_BYTES:
                raise HTTPException(413, "file too large")
            out.write(chunk)
    return {"filename": file.filename, "size": written, "note": note}

@app.post("/upload-many")
async def upload_many(files: list[UploadFile]):
    return [f.filename for f in files]

# pip install python-multipart  <- otherwise RuntimeError at startup

Answer

FastAPI supports both. A `def` (sync) handler runs in a threadpool, useful for blocking code like synchronous database drivers (psycopg2). An `async def` handler runs in the main event loop, required for `await`ing async code (httpx, asyncpg, SQLAlchemy async).

The rule: if your handler awaits anything, it must be async; if it does CPU-bound work, sync is fine. Mixing is the most common FastAPI bug, calling a blocking function inside an `async def` blocks the whole event loop. The detail that separates a junior from a senior answer is the threadpool itself: FastAPI hands sync handlers to AnyIO's worker thread pool, which defaults to 40 threads per process.

So a `def` endpoint that takes 2 seconds caps that worker at roughly 20 requests per second no matter how many clients connect, and request 41 waits for a free thread rather than erroring. You raise that ceiling with `anyio.to_thread.current_default_thread_limiter().total_tokens = 100`, though at some point more workers beat more threads. The failure signature of the opposite mistake, blocking inside `async def`, is distinctive: p99 latency on completely unrelated endpoints spikes together, health checks time out, and CPU sits low.

Common culprits are `requests.get`, `time.sleep`, `psycopg2`, `boto3`, `open().read()` on a network mount, and `bcrypt.hashpw`. The fixes are an async client (`httpx.AsyncClient`, `asyncpg`, `aioboto3`), `await run_in_threadpool(fn)` from `starlette.concurrency`, or `await asyncio.to_thread(fn)`. Dependencies follow the same rule independently of the handler: a `def` dependency is threadpooled even when the route is `async def`.

import time
import httpx
from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool

app = FastAPI()

# WRONG: blocking call inside async def stalls every other request
@app.get("/bad")
async def bad():
    time.sleep(2)
    return {"ok": True}

# OK: sync handler, FastAPI runs it in the threadpool
@app.get("/sync-ok")
def sync_ok():
    time.sleep(2)
    return {"ok": True}

# BEST: real async I/O
@app.get("/good")
async def good():
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com/ping", timeout=2.0)
    return r.json()

# escape hatch for a blocking lib inside async def
@app.get("/escape")
async def escape():
    return await run_in_threadpool(legacy_blocking_call)

Answer

Use the `Header()` dependency. FastAPI converts `-` to `_` in header names by default (so `X-Token` becomes the parameter `x_token`). You can disable this with `convert_underscores=False`, which you need for the rare header that genuinely contains an underscore.

Header matching is case-insensitive because HTTP header names are, so `x-token`, `X-Token` and `X-TOKEN` all bind to the same parameter. A header declared without a default is required and a missing one produces a 422 with `loc` of `["header", "x-token"]`, which is a cleaner contract than reading `request.headers.get(...)` and hand-rolling the error. Repeated headers bind to `list[str]`, which matters for `X-Forwarded-For` chains and multiple `Accept` values.

When you need the raw mapping, accept `request: Request` and read `request.headers`, a case-insensitive multidict. Production points a senior interviewer will push on: behind a load balancer, `request.client.host` is the proxy, so the real client IP comes from `X-Forwarded-For` and you must run Uvicorn with `--proxy-headers --forwarded-allow-ips="*"` (or the specific proxy IP) for `X-Forwarded-Proto` to be honoured, otherwise generated URLs come out as http behind an https load balancer. Never trust a forwarded header that your own edge did not set. `Cookie()` works the same way for cookies, and for auth you normally prefer the `fastapi.security` classes over a raw `Header()` because they also register the scheme in the OpenAPI spec and light up the Authorize button in Swagger UI.

from typing import Annotated
from fastapi import Cookie, FastAPI, Header, Request

app = FastAPI()

@app.get("/profile")
async def profile(
    x_token: Annotated[str, Header()],                      # required -> 422 if absent
    user_agent: Annotated[str | None, Header()] = None,     # User-Agent
    x_forwarded_for: Annotated[list[str], Header()] = [],
    session: Annotated[str | None, Cookie()] = None,
):
    return {"token": x_token, "ua": user_agent, "xff": x_forwarded_for, "session": session}

@app.get("/raw")
async def raw(request: Request):
    return dict(request.headers)

# uvicorn main:app --proxy-headers --forwarded-allow-ips="10.0.0.0/8"

Answer

FastAPI's `Depends()` system lets you declare reusable functions whose return values are injected into route handlers. Common uses: shared database connections, auth checks, query parameter parsing, feature flags. The dependency is itself a function, FastAPI calls it, awaits its result, and passes it in.

Dependencies can have dependencies, forming a tree FastAPI resolves per request. Unlike Spring or NestJS containers there is no global registry and no singleton scope: resolution is per request, driven entirely by the function signature, and the dependency's own parameters are parsed from the request exactly like a handler's, so a dependency can itself declare headers, query params or another `Depends`. A dependency can be a plain function, an `async def`, a class (FastAPI calls `__init__` with the parsed params), or a generator using `yield`, which is how you get setup and teardown: code after `yield` runs once the response has been generated, which is where you close sessions or release locks.

Three behaviours interviewers probe. First, results are cached per request, so two dependencies that both need `get_current_user` call it once. Second, teardown ordering is reverse of setup, and an exception in the handler propagates into the generator so you can roll back a transaction in an `except` block. Third, dependencies that return nothing but must still run (audit logging, permission checks) attach via `dependencies=[Depends(...)]` on the route, router, or the `FastAPI(dependencies=[...])` constructor for app-wide enforcement. `app.dependency_overrides[get_db] = fake_db` is the same machinery, which is what makes FastAPI so testable.

from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException

app = FastAPI()

async def get_db():
    db = await create_connection()
    try:
        yield db          # handler runs here
    except Exception:
        await db.rollback()
        raise
    finally:
        await db.close()  # always runs, after the response is generated

async def require_admin(x_role: Annotated[str, Header()]):
    if x_role != "admin":
        raise HTTPException(403, "admin only")

DB = Annotated[object, Depends(get_db)]

# no value injected, the dependency just has to pass
@app.get("/items", dependencies=[Depends(require_admin)])
async def list_items(db: DB):
    return await db.fetch_all("SELECT * FROM items")

Answer

Use the `status_code` parameter on the route decorator for the default status. For dynamic status or headers, accept `Response` as a parameter and mutate it. For one-off responses, return a `JSONResponse` or `Response` object directly with the status and headers you want.

Three approaches with different trade-offs. The decorator argument is declarative and shows up in the OpenAPI spec, so `status_code=201` on a create endpoint documents itself. Injecting `response: Response` lets you set `response.status_code` and `response.headers[...]` while still returning a plain object, which keeps `response_model` filtering active, this is the right choice when you need a header like `X-Total-Count` alongside a validated body.

Returning a `JSONResponse` yourself gives full control but bypasses `response_model` entirely: nothing is filtered, nothing is validated, and the OpenAPI schema no longer matches reality unless you also declare `responses={404: {...}}`. For error paths prefer `raise HTTPException(status_code=404, detail="not found", headers={"X-Error": "missing"})` over returning an error response, because raising unwinds the dependency `yield` teardown correctly and produces the standard `{"detail": ...}` shape. Details that catch people out: 204 and 304 must have an empty body, returning content with 204 makes some proxies and browsers error, so use `Response(status_code=204)`; and a route declared `status_code=201` still returns 200 if you hand back your own `JSONResponse` without repeating the status. `fastapi.status` exists purely so `status.HTTP_422_UNPROCESSABLE_ENTITY` autocompletes instead of a bare integer.

from fastapi import FastAPI, HTTPException, Response, status
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item():
    return {"id": 1}

@app.get("/items")
async def list_items(response: Response):
    rows = await db.fetch_all("SELECT * FROM items LIMIT 50")
    response.headers["X-Total-Count"] = str(await db.count())
    return rows  # response_model filtering still applies

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    row = await db.fetch_item(item_id)
    if row is None:
        raise HTTPException(status.HTTP_404_NOT_FOUND, detail="item not found")
    return row

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    await db.delete_item(item_id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)  # empty body

# raw control, but response_model no longer filters anything
@app.get("/legacy")
async def legacy():
    return JSONResponse(status_code=418, content={"error": "teapot"})

Answer

FastAPI introspects every route handler at startup: it reads the type hints, response_model, path/query/body params, status codes, and docstrings to build a complete OpenAPI 3.1 spec. The spec is served at `/openapi.json` and rendered as Swagger UI at `/docs` and ReDoc at `/redoc`. You can add tags, summaries, and examples via decorator arguments.

For production, you can disable the docs (`FastAPI(docs_url=None)`) or protect them behind auth. Details worth knowing: the schema is built lazily on the first request to `/openapi.json` and then cached on `app.openapi_schema`, so mutating routes at runtime will not refresh it unless you clear that attribute. The first line of the docstring becomes the summary and the rest becomes the description, and Markdown renders. `operation_id` defaults to a long generated name, which produces ugly method names in clients generated by `openapi-typescript-codegen` or `openapi-generator`, so teams that generate a frontend SDK either set `operation_id` explicitly or post-process names from the route name.

Since FastAPI 0.99 the output is OpenAPI 3.1 with proper JSON Schema 2020-12, which matters because older tooling pinned to 3.0 chokes on `"type": ["string", "null"]` for optional fields. `include_in_schema=False` hides internal routes such as health checks. A common production setup is to keep `openapi_url` enabled but gate `/docs` behind HTTP Basic or an internal-only ingress, and to commit the generated `openapi.json` to CI so a breaking API change shows up as a diff in code review rather than in a client outage.

from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi.security import HTTPBasic, HTTPBasicCredentials
import secrets

app = FastAPI(title="Payments API", version="2.3.0", docs_url=None, redoc_url=None)
basic = HTTPBasic()

def docs_auth(creds: HTTPBasicCredentials = Depends(basic)):
    if not secrets.compare_digest(creds.password, DOCS_PASSWORD):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, headers={"WWW-Authenticate": "Basic"})

@app.get("/docs", include_in_schema=False, dependencies=[Depends(docs_auth)])
async def protected_docs():
    return get_swagger_ui_html(openapi_url="/openapi.json", title="Payments API")

@app.get("/healthz", include_in_schema=False)
async def healthz():
    return {"status": "ok"}

@app.post("/refunds", tags=["payments"], operation_id="createRefund", status_code=201)
async def create_refund():
    """Create a refund.

    Refunds settle in **3-5 working days**.
    """
    return {"id": "rfnd_1"}

Answer

Use `app.mount()` with `StaticFiles` from `fastapi.staticfiles`. This mounts a directory under a URL prefix. For most production setups, you'd front the API with nginx or CloudFront and serve static files there instead, `StaticFiles` is convenient for local dev or low-traffic admin assets.

Mechanically, `mount` attaches a whole sub-application to a path prefix, so everything under `/static` leaves FastAPI's routing and never appears in the OpenAPI schema. `StaticFiles` is a real file server: it sets `Content-Type` from the extension, emits `Last-Modified` and `ETag`, answers conditional requests with 304, and supports HTTP range requests so video seeking works. Reads go through the threadpool, which is why a busy static mount competes with your `def` endpoints for the same 40 worker threads, the concrete reason to move assets to a CDN. Practical points: `StaticFiles(directory="static", html=True)` serves `index.html` for directory paths and is the usual way to host a built React or Vite SPA behind the same origin as the API, though you still need a catch-all route for client-side routes so a refresh on `/dashboard/settings` does not 404.

Mount order matters, mounts are checked in registration order, so mounting at `/` first shadows every API route declared afterwards; mount the SPA last. The directory must exist at startup or you get `RuntimeError: Directory 'static' does not exist`, which is a frequent Docker failure when the build stage does not copy the folder. There is no cache-control header by default, so add one with middleware or fingerprinted filenames.

from fastapi import FastAPI
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

app = FastAPI()

@app.get("/api/health")
async def health():
    return {"ok": True}

# assets first
app.mount("/assets", StaticFiles(directory="dist/assets"), name="assets")

# SPA fallback LAST, otherwise it shadows /api/*
app.mount("/", StaticFiles(directory="dist", html=True), name="spa")

# alternative: explicit catch-all so deep links resolve
@app.get("/{full_path:path}", include_in_schema=False)
async def spa_fallback(full_path: str):
    return FileResponse("dist/index.html")

Answer

CORS (Cross-Origin Resource Sharing) is a browser security mechanism, by default, JS on `app.com` can't fetch from `api.app.com`. FastAPI ships with `CORSMiddleware` to handle preflight (OPTIONS) requests and set the right `Access-Control-*` headers. In production, never use `allow_origins=['*']` if you also use `allow_credentials=True`, the spec forbids the literal `*` wildcard alongside `Access-Control-Allow-Credentials: true`, so browsers reject that combination outright.

Starlette's `CORSMiddleware` papers over it: when the request carries a cookie it echoes the requesting origin back in `Access-Control-Allow-Origin` instead of sending `*`, which means the wildcard you thought was a dev convenience quietly becomes allow-any-origin-with-credentials, while cookie-less credentialed requests still get `*` and are blocked. Use an explicit origin list, or `allow_origin_regex` for preview deployments. What actually happens on the wire: a cross-origin request that is not a simple GET/POST with a basic content type triggers a preflight `OPTIONS` carrying `Access-Control-Request-Method` and `Access-Control-Request-Headers`; `CORSMiddleware` answers it directly without ever reaching your route, so a preflight 400 usually means the method or header is not in your allow lists.

Origins must match scheme, host and port exactly, `https://app.example.com` does not cover `http://app.example.com` or a trailing slash. Two more traps: browsers only expose a handful of response headers to JS, so a custom `X-Total-Count` needs `expose_headers=["X-Total-Count"]`; and CORS is browser-only enforcement, curl, Postman and a server-side proxy ignore it entirely, so it is never an authorisation control. `max_age=600` caches the preflight and removes a round trip per unique request shape.

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

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com", "http://localhost:5173"],
    allow_origin_regex=r"https://.*\.preview\.example\.com",
    allow_credentials=True,                      # never combine with allow_origins=["*"]
    allow_methods=["GET", "POST", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    expose_headers=["X-Total-Count", "X-Request-ID"],
    max_age=600,                                 # cache preflight for 10 minutes
)

Answer

WSGI (PEP 3333) is the synchronous Python web server interface used by Flask and classic Django: the server calls `application(environ, start_response)` once per request, that call blocks a worker until the response is complete, and there is no place in the contract for long-lived connections. ASGI is the async successor. An ASGI app is `async def app(scope, receive, send)`, where `scope` is a dict describing the connection (`scope["type"]` is `http`, `websocket` or `lifespan`), `receive` is an awaitable that yields incoming events such as `http.request` body chunks, and `send` pushes out `http.response.start` and `http.response.body` events.

Because the request and response are event streams rather than one blocking call, a single worker can hold thousands of open connections while awaiting I/O, and the same interface covers WebSockets, server-sent events and startup/shutdown (the `lifespan` scope). FastAPI is an ASGI framework via Starlette, so it needs an ASGI server: Uvicorn (uvloop plus httptools), Hypercorn (adds HTTP/2 and HTTP/3), or Granian. Running it under Gunicorn alone fails because Gunicorn is a WSGI server, which is why the deployment recipe is `-k uvicorn.workers.UvicornWorker`.

The practical consequence interviewers are testing for: ASGI buys concurrency, not parallelism. One event loop still runs one Python bytecode stream, so CPU-bound work blocks everything and you still scale processes across cores. `a2wsgi` bridges the two directions when you need to mount a legacy Flask or Django app inside FastAPI.

# the raw ASGI contract that Starlette (and so FastAPI) implements
async def app(scope, receive, send):
    assert scope["type"] == "http"
    await receive()  # http.request event(s), body chunks
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [(b"content-type", b"application/json")],
    })
    await send({"type": "http.response.body", "body": b'{"ok":true}'})

# mounting a legacy WSGI app inside an ASGI one
from fastapi import FastAPI
from a2wsgi import WSGIMiddleware
from legacy_flask_app import flask_app

api = FastAPI()
api.mount("/legacy", WSGIMiddleware(flask_app))

Answer

JSON bodies map to Pydantic models, but `application/x-www-form-urlencoded` and `multipart/form-data` need `Form()` because the parsing path is different. Declaring `username: Annotated[str, Form()]` tells FastAPI to pull the field from the parsed form rather than from JSON, and like file uploads it requires the `python-multipart` package installed, otherwise the app raises a `RuntimeError` at import. A single endpoint cannot mix a JSON body with `Form()` fields: the moment any parameter is a form field, the whole body is parsed as a form, so a `BaseModel` parameter alongside it will not receive anything.

FastAPI 0.113 added form models, so you can group fields into a Pydantic model with `Annotated[LoginForm, Form()]` and get validation and (in 0.114+) `extra="forbid"` rejection of unexpected fields, which is much cleaner than ten loose parameters. `OAuth2PasswordRequestForm` from `fastapi.security` is a ready-made dependency for the OAuth2 password grant: it declares the spec-mandated `username`, `password`, `grant_type`, `scope`, `client_id` and `client_secret` form fields, exposes `scopes` as a parsed list, and pairs with `OAuth2PasswordBearer(tokenUrl="/token")` so the Authorize button in Swagger UI actually works end to end. The field is called `username` even when your users log in with an email or phone number, that is the OAuth2 spec, not FastAPI, and renaming it breaks generated clients. Return `{"access_token": ..., "token_type": "bearer"}` exactly, because that is what the spec and the Swagger UI client expect.

from typing import Annotated
from fastapi import Depends, FastAPI, Form, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from pydantic import BaseModel

app = FastAPI()

@app.post("/token")
async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()]):
    user = await authenticate(form.username, form.password)
    if not user:
        raise HTTPException(400, "Incorrect username or password")
    return {"access_token": make_jwt(user.id, form.scopes), "token_type": "bearer"}

# form model, FastAPI 0.113+
class ContactForm(BaseModel):
    name: str
    email: str
    message: str

@app.post("/contact")
async def contact(data: Annotated[ContactForm, Form()]):
    return {"received": data.email}

# curl -X POST /token -d "username=a@b.co&password=secret"

Answer

`APIRouter` is a mini `FastAPI` app you define per domain module and attach with `app.include_router(...)`. The usual layout is `app/main.py` for app creation, `app/api/routers/users.py`, `orders.py` and so on, `app/models` for SQLAlchemy, `app/schemas` for Pydantic, `app/services` for business logic and `app/deps.py` for shared dependencies. `include_router` accepts `prefix` (must start with a slash and must not end with one), `tags` for docs grouping, `dependencies` applied to every route in the router, and `responses` for shared error documentation. This is where you enforce auth once per module rather than repeating `Depends(current_user)` on twenty endpoints.

Routers nest, so `api_v1 = APIRouter(prefix="/api/v1")` can include the users and orders routers and the app includes just `api_v1`, which is how versioning is normally done. Practical details: the same router can be included twice under different prefixes, but every route needs a unique `operation_id`, otherwise generated clients collide; circular imports are the number one problem in this layout and the fix is to keep `deps.py` free of router imports and to import models lazily inside functions when a service needs a router-level type. Route order still applies inside a router. Interviewers often ask where business logic lives: the honest answer is not in the router, handlers should parse, delegate to a service function that knows nothing about HTTP, and shape the response, which is what makes the logic testable without a `TestClient`.

# app/api/routers/users.py
from fastapi import APIRouter, Depends
from app.deps import current_user

router = APIRouter(
    prefix="/users",
    tags=["users"],
    dependencies=[Depends(current_user)],       # applies to every route below
    responses={404: {"description": "Not found"}},
)

@router.get("", operation_id="listUsers")
async def list_users():
    return await user_service.list_all()

# app/main.py
from fastapi import FastAPI, APIRouter
from app.api.routers import users, orders

api_v1 = APIRouter(prefix="/api/v1")
api_v1.include_router(users.router)
api_v1.include_router(orders.router)

app = FastAPI()
app.include_router(api_v1)
# -> GET /api/v1/users

Answer

Standard pattern: a login endpoint validates credentials and returns a signed JWT; protected endpoints use a dependency that extracts and verifies the token from the Authorization header. `OAuth2PasswordBearer` from `fastapi.security` handles the Bearer token extraction and registers the scheme in OpenAPI so Swagger UI gets an Authorize button. Store the JWT secret in env vars (never in code). Refresh tokens go in HttpOnly cookies to limit XSS exposure; access tokens are short-lived (15-30 minutes).

Library note for 2026: `python-jose` has been effectively unmaintained, and most teams have moved to `PyJWT` or `joserfc`, so quoting `pip install pyjwt[crypto]` reads as current. Verification details a senior interviewer will dig into: always pass an explicit `algorithms=["HS256"]` list, because accepting the token's own `alg` header enables the classic `alg: none` and RS256-to-HS256 confusion attacks; validate `exp`, `iat`, and also `aud` and `iss` when multiple services share a key; and use `secrets.compare_digest` rather than `==` anywhere you compare secrets. HS256 with a shared secret is fine inside one service, but the moment several services verify tokens, switch to RS256 or EdDSA so only the issuer holds the private key and everyone else fetches public keys from a JWKS endpoint.

The real weakness of JWT is revocation: a stolen access token stays valid until it expires, so keep the TTL short, store refresh tokens server-side with rotation and reuse detection, and keep a Redis deny-list keyed on the `jti` claim for forced logout. Password hashing belongs to `argon2` or `bcrypt` and, because both are deliberately CPU-heavy, must not run inline in an `async def` handler.

import os, uuid
from datetime import datetime, timedelta, timezone
from typing import Annotated
import jwt  # pip install pyjwt[crypto]
from jwt.exceptions import InvalidTokenError
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer

SECRET = os.environ["JWT_SECRET"]
app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl="/token")

def make_token(user_id: str) -> str:
    now = datetime.now(timezone.utc)
    return jwt.encode(
        {"sub": user_id, "jti": str(uuid.uuid4()), "iat": now,
         "exp": now + timedelta(minutes=15), "aud": "payments", "iss": "auth-svc"},
        SECRET, algorithm="HS256",
    )

async def current_user(token: Annotated[str, Depends(oauth2)]) -> str:
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"],
                             audience="payments", issuer="auth-svc")
    except InvalidTokenError:
        raise HTTPException(401, "Invalid token", headers={"WWW-Authenticate": "Bearer"})
    if await redis.exists(f"revoked:{payload['jti']}"):
        raise HTTPException(401, "Token revoked")
    return payload["sub"]

@app.get("/me")
async def me(user_id: Annotated[str, Depends(current_user)]):
    return {"user_id": user_id}

Answer

Two mature patterns in 2026: SQLAlchemy 2.0 (async) for full ORM features, or SQLModel (same author as FastAPI) for simpler Pydantic-integrated models. Wire the engine and session as a FastAPI dependency so each request gets its own session, never share a session across requests because of how SQLAlchemy's identity map works and because `AsyncSession` is not safe for concurrent use. Production: connection pool size 5-20 (match your concurrency), use PgBouncer with PostgreSQL if you have 100+ workers.

The parts people get wrong: the pool lives per process, so real database concurrency is `(pool_size + max_overflow) * workers`, and four Gunicorn workers with `pool_size=20` quietly opens up to 80 connections against a Postgres box whose `max_connections` may be 100. `expire_on_commit=False` matters because the default expires every attribute after commit, so touching `user.email` after the commit fires a lazy refresh that raises `MissingGreenlet` in async code. That same error is the signature of lazy loading in general: in async SQLAlchemy you must eager-load relationships with `selectinload` or `joinedload`, or use `AsyncAttrs` and `await obj.awaitable_attrs.items`. Use `postgresql+asyncpg://` for async and `postgresql+psycopg2://` for sync; mixing a sync driver into an async engine fails at connect time.

If you front Postgres with PgBouncer in transaction mode, disable statement caching (`connect_args={"statement_cache_size": 0}`) and set `poolclass=NullPool`, otherwise asyncpg's prepared statements break across pooled connections. Commit explicitly in the handler or the service, and keep the rollback in the dependency's `except` so a raised `HTTPException` never leaves a half-written transaction.

from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import selectinload
from sqlalchemy import select

app = FastAPI()
engine = create_async_engine(
    "postgresql+asyncpg://user:pw@db/app",
    pool_size=10, max_overflow=5, pool_pre_ping=True, pool_recycle=1800,
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

DB = Annotated[AsyncSession, Depends(get_db)]

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: DB):
    stmt = (select(User)
            .options(selectinload(User.orders))   # avoids MissingGreenlet
            .where(User.id == user_id))
    return (await db.execute(stmt)).scalar_one_or_none()

Answer

FastAPI's `BackgroundTasks` runs a function AFTER the response is sent, useful for fire-and-forget operations like sending emails, logging, or analytics. Limitation: the task runs in the same process, so it dies if the app restarts. For anything where loss-on-restart is unacceptable (payment confirmations, critical workflows), use a real queue: Celery, RQ, or Dramatiq.

Payment flows built on Razorpay-style webhooks almost always use Celery with Redis for exactly this reason. Behaviour that surprises people: the task is not detached, it runs inside the same ASGI response cycle, so the client already has the complete response body while the worker and its connection slot stay tied up until the task finishes, which silently eats serving capacity under load rather than showing up as slow responses. A sync (`def`) task goes to the threadpool, an `async def` task runs on the loop and will block it if it does anything blocking.

Exceptions inside a background task never reach the client, the response has already gone out, so wrap the body in try/except and log, otherwise failures are invisible. There is no retry, no backoff, no visibility and no deduplication. The dependency interaction is the classic bug: a database session yielded by `Depends(get_db)` is torn down before background tasks run in older versions and its teardown ordering has shifted between releases, so passing the session into `add_task` gives you a closed session.

Pass primitive IDs instead and open a fresh session inside the task. Rule of thumb for interviews: `BackgroundTasks` for best-effort, sub-second, idempotent work; a broker with an outbox table for anything a customer would complain about losing.

import logging
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()
log = logging.getLogger(__name__)

async def send_welcome_email(user_id: int) -> None:
    # open your own session/client here, never reuse the request's session
    try:
        async with SessionLocal() as db:
            user = await db.get(User, user_id)
            await mailer.send(user.email, "Welcome!")
    except Exception:
        log.exception("welcome email failed user_id=%s", user_id)  # silent otherwise

@app.post("/signup", status_code=201)
async def signup(email: str, background_tasks: BackgroundTasks):
    user = await create_user(email)
    background_tasks.add_task(send_welcome_email, user.id)  # pass an id, not the session
    return {"created": user.id}

# durable alternative when loss is unacceptable
# celery_app.send_task("tasks.send_receipt", args=[payment_id], retry=True)

Answer

Register exception handlers with `@app.exception_handler(ExceptionType)`. Common targets: custom domain exceptions (e.g. `UserNotFoundError`), and Pydantic's `RequestValidationError` to customize the 422 response format. Avoid catching `Exception` broadly, let unexpected errors bubble to FastAPI's default 500 handler, which is wired to logging.

How dispatch works: Starlette walks the exception's MRO and picks the most specific registered handler, so registering a handler for a base `DomainError` covers every subclass, and you can also register by status code with `@app.exception_handler(404)`. Overriding `RequestValidationError` is the usual first customisation because the default `detail` array does not match most frontend contracts; `exc.errors()` gives you the structured list and `exc.body` the raw payload, and remember to keep returning 422 unless you have a real reason to switch to 400. `HTTPException` is handled separately by `http_exception_handler`, and `StarletteHTTPException` is its parent, so a handler for the Starlette class also catches 404s raised by the router itself, which is how you make unmatched routes return your JSON error shape instead of the default. Three production notes.

Exception handlers do not run for errors raised inside `BaseHTTPMiddleware` or in a `StreamingResponse` generator after the first byte has been flushed, because the status line is already committed. Never put `exc` or a traceback into the response body, log it with the request ID and return an opaque message. And if you use Sentry or OpenTelemetry, a catch-all `Exception` handler swallows the error before the integration sees it unless you re-report it explicitly.

import logging
from fastapi import FastAPI, Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()
log = logging.getLogger(__name__)

class DomainError(Exception):
    status = 400
    code = "domain_error"

class NotFound(DomainError):
    status = 404
    code = "not_found"
    def __init__(self, resource: str): self.resource = resource

@app.exception_handler(DomainError)          # also catches NotFound via the MRO
async def domain_handler(request: Request, exc: DomainError):
    return JSONResponse(exc.status, {"code": exc.code, "message": str(exc)})

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    fields = [{"field": ".".join(map(str, e["loc"][1:])), "error": e["msg"]} for e in exc.errors()]
    return JSONResponse(status.HTTP_422_UNPROCESSABLE_ENTITY, {"code": "invalid_request", "fields": fields})

@app.exception_handler(StarletteHTTPException)  # covers router 404/405 too
async def http_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(exc.status_code, {"code": "http_error", "message": exc.detail})

Answer

`Annotated[X, Depends(func)]` (PEP 593, Python 3.9+) is the modern, recommended form. It's reusable, you can define a type alias and reuse it across all routes. Functionally identical to `x: X = Depends(func)` but reads cleaner and supports multiple metadata items per parameter.

FastAPI 0.95+ documentation uses `Annotated` exclusively. The differences are not only cosmetic. With the default-value form, the parameter has a default, so it must come after all positional parameters, which forces awkward ordering once several dependencies are involved, and calling the function directly in a unit test passes a `Depends` object rather than a real value. `Annotated` keeps the parameter required from Python's point of view, so ordering is free and a direct call raises a clear `TypeError` instead of silently handing you a sentinel.

Static analysis is the bigger win: mypy and Pyright see `user: CurrentUser` as a `User`, whereas `user: User = Depends(current_user)` reads to the type checker as a default of the wrong type and needs a cast or an ignore. `Annotated` also stacks metadata, so `Annotated[int, Query(ge=1), Depends(validate_page)]` is expressible while the old form allows exactly one. The same syntax covers `Query`, `Path`, `Header`, `Cookie`, `Form`, `File` and `Body`, and Pydantic `Field` constraints compose inside it. The practical payoff in a real codebase is a `deps.py` holding `CurrentUser`, `DB`, `Pagination` and `TenantId` aliases, so a handler signature documents its whole dependency graph in one line and swapping an implementation is a one-line change.

from typing import Annotated
from fastapi import Depends, FastAPI, Query
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

# app/deps.py, reused across every router
DB = Annotated[AsyncSession, Depends(get_db)]
CurrentUser = Annotated[User, Depends(current_user)]
Page = Annotated[int, Query(ge=1, le=1000)]

# old form: defaults force ordering, type checkers complain
@app.get("/legacy")
async def legacy(db: AsyncSession = Depends(get_db), page: int = Query(1, ge=1)):
    return await db.list(page)

# Annotated form: no defaults needed, params can be in any order
@app.get("/me")
async def me(user: CurrentUser, db: DB, page: Page = 1):
    return {"user": user.id, "page": page}

# unit-testable without the HTTP layer
# await me(user=fake_user, db=fake_session, page=1)

Answer

Use `TestClient` for most cases or `httpx.AsyncClient` if your tests need to await async setup. `TestClient` is built on httpx (it was on `requests` before Starlette 0.21) and drives the ASGI app in-process, so no server, no port, no network. Best practice: use `app.dependency_overrides` to replace database and external-service dependencies with test fakes, and clear the dict in a fixture teardown so overrides do not leak between test modules. Things that decide whether the answer sounds experienced. `TestClient` only runs `lifespan` (startup and shutdown) when used as a context manager, `with TestClient(app) as client:`, so a test that fails with an uninitialised connection pool is usually missing that `with`.

Overriding by the function object means importing the exact same object the route imported; a duplicate import path gives you a key that never matches and the real dependency runs against your production database. For a genuine database, the fast pattern is one transaction per test: open a connection, begin a nested transaction, hand the session to the override, and roll back afterwards, which is far quicker than recreating schema per test. External HTTP calls get stubbed with `respx` for httpx, or by injecting a client through a dependency and overriding it.

Test the contract, not the framework: assert status codes, the 422 `detail` shape for bad input, authorisation failures, and that a response never contains fields your public model omits. Run with `pytest -x --cov=app` in CI, and use `parametrize` for validation matrices instead of copying assertions.

import pytest
from fastapi.testclient import TestClient
from main import app, get_db

@pytest.fixture
def client():
    async def fake_db():
        yield InMemoryDB()
    app.dependency_overrides[get_db] = fake_db
    with TestClient(app) as c:   # `with` runs lifespan startup/shutdown
        yield c
    app.dependency_overrides.clear()

def test_create_user(client):
    r = client.post("/users", json={"email": "a@b.co", "age": 30, "full_name": "A"})
    assert r.status_code == 201
    assert "password_hash" not in r.json()

@pytest.mark.parametrize("payload,loc", [
    ({"email": "nope", "age": 30, "full_name": "A"}, ["body", "email"]),
    ({"email": "a@b.co", "age": 5, "full_name": "A"}, ["body", "age"]),
])
def test_validation(client, payload, loc):
    r = client.post("/users", json=payload)
    assert r.status_code == 422
    assert r.json()["detail"][0]["loc"] == loc

Answer

FastAPI doesn't ship rate limiting natively. Three production options: (1) `slowapi`, which wraps the `limits` package and is easy to wire as a decorator plus exception handler, (2) `fastapi-limiter`, a Redis-backed dependency, (3) a custom dependency running a Lua script against Redis. For multi-instance deployments, you MUST use a shared store (Redis), in-memory counters don't work behind a load balancer and they also reset on every deploy.

Typical limits: 60 requests per minute per IP for unauthenticated endpoints and 600 per minute per user for authenticated ones, with much tighter caps (5 per minute) on login, OTP send and password reset. Algorithm choice is what interviewers actually probe. A fixed window is one `INCR` with an `EXPIRE` and is cheap, but it allows a double burst across the boundary.

A sliding window log in a Redis sorted set (`ZREMRANGEBYSCORE` then `ZADD` then `ZCARD`) is exact but stores one member per request. A token bucket allows controlled bursts and is the usual choice for paid API tiers. All three must be atomic, so run them as a single Lua script or a `MULTI` pipeline, otherwise concurrent workers race and overshoot the limit.

Keying matters as much as counting: keying on IP alone punishes everyone behind one corporate NAT or a mobile carrier CGNAT, so authenticated traffic should key on user or API key and fall back to IP. Always return 429 with `Retry-After` plus `X-RateLimit-Limit` and `X-RateLimit-Remaining`, and decide explicitly whether Redis being down should fail open or fail closed. At scale, push the coarse limit into nginx, Cloudflare or the API gateway and keep the application layer for per-user business quotas.

import time
from fastapi import Depends, FastAPI, HTTPException, Request, Response

app = FastAPI()

# atomic fixed-window counter: INCR + EXPIRE in one round trip
LUA = """
local c = redis.call('INCR', KEYS[1])
if c == 1 then redis.call('EXPIRE', KEYS[1], ARGV[1]) end
return c
"""

def rate_limit(limit: int, window: int):
    async def dep(request: Request, response: Response):
        ident = getattr(request.state, "user_id", None) or request.client.host
        bucket = int(time.time()) // window
        key = f"rl:{request.url.path}:{ident}:{bucket}"
        count = await redis.eval(LUA, 1, key, window)
        response.headers["X-RateLimit-Limit"] = str(limit)
        response.headers["X-RateLimit-Remaining"] = str(max(0, limit - count))
        if count > limit:
            raise HTTPException(429, "Too Many Requests",
                                headers={"Retry-After": str(window)})
    return dep

@app.post("/auth/otp", dependencies=[Depends(rate_limit(5, 60))])
async def send_otp(phone: str):
    return {"sent": True}

Answer

Middleware runs around EVERY request, both before (modifying the request) and after (modifying the response). Dependencies run only for routes that declare them, and dependencies with `yield` handle teardown but cannot alter the response object. Use middleware for cross-cutting concerns: CORS, request logging, GZIP compression, request IDs.

Use dependencies for auth, DB sessions, and per-route validation. Deeper differences worth naming. Middleware sits outside the router, so it also sees requests that never match a route, which is exactly why 404s and 405s still get a request ID and a log line while a dependency would never fire.

It cannot raise `HTTPException` usefully, because that is a router-level concept, so a middleware rejection returns a `JSONResponse` or `PlainTextResponse` directly. Middleware cannot inject values into a handler signature either, it passes state through `request.state`, which is untyped and invisible to the type checker, whereas a dependency returns a typed object. Ordering catches people out: `add_middleware` prepends, so the last one added is the outermost layer that sees the request first and the response last.

The `@app.middleware("http")` decorator builds a `BaseHTTPMiddleware`, which wraps the response in an anyio task and therefore breaks `StreamingResponse` back-pressure and swallows exceptions before your `@app.exception_handler` sees them; for hot paths, write pure ASGI middleware instead. `GZipMiddleware(minimum_size=1000)` is the standard compression layer, and it must be outside anything that sets `Content-Length`. Performance: every middleware runs on every request including health checks, so three chatty middlewares can cost more than the handler.

import time, uuid
from fastapi import FastAPI, Request
from fastapi.middleware.gzip import GZipMiddleware

app = FastAPI()

@app.middleware("http")
async def request_id_middleware(request: Request, call_next):
    rid = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    request.state.request_id = rid
    start = time.perf_counter()
    response = await call_next(request)   # also runs for 404s and 405s
    response.headers["X-Request-ID"] = rid
    response.headers["X-Process-Time"] = f"{time.perf_counter() - start:.4f}"
    return response

# added last => outermost: compresses everything the layers below produce
app.add_middleware(GZipMiddleware, minimum_size=1000)

# pure ASGI middleware: no BaseHTTPMiddleware overhead, streaming-safe
class TimingASGIMiddleware:
    def __init__(self, app): self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            return await self.app(scope, receive, send)
        start = time.perf_counter()
        async def send_wrapper(message):
            if message["type"] == "http.response.start":
                message["headers"].append(
                    (b"x-elapsed", f"{time.perf_counter() - start:.4f}".encode()))
            await send(message)
        await self.app(scope, receive, send_wrapper)

Answer

Return a `StreamingResponse` with an async generator. The generator yields bytes or str, and Starlette sends each chunk as it arrives using HTTP chunked transfer encoding, so no `Content-Length` is set and memory stays flat regardless of payload size. Common uses: CSV exports of millions of rows, video streaming, proxying an object from S3, and token-by-token LLM output.

For SSE specifically, set `media_type='text/event-stream'` and yield `data: {...}\n\n` formatted messages. The part that decides whether it actually works in production is everything between your process and the browser. `GZipMiddleware` buffers, so exclude streaming routes or you lose incrementality. nginx buffers proxied responses by default, so send `X-Accel-Buffering: no` or set `proxy_buffering off`. An AWS ALB is fine but idle timeout (60s default) will cut a quiet stream, which is why long-lived streams send a heartbeat comment every 15 to 30 seconds.

Inside the generator you must not hold a database session open for the whole export if the pool is small, use a server-side cursor with `stream_results=True` or paginate by keyset. Client disconnects are the other trap: the generator keeps running after the browser goes away unless you check `await request.is_disconnected()`, and an uncaught exception mid-stream cannot change the status code because 200 has already been flushed, so callers see a truncated body instead of an error. For downloads, add `Content-Disposition: attachment; filename=...`, and prefer `FileResponse` over a hand-written generator when the bytes are already a file on disk, since it uses sendfile.

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()

async def csv_rows(request: Request):
    yield "id,name,email\n"
    async with engine.connect() as conn:
        result = await conn.stream(select(User.id, User.name, User.email))
        async for row in result:
            if await request.is_disconnected():
                return              # stop burning CPU on an abandoned download
            yield f"{row.id},{row.name},{row.email}\n"

@app.get("/users.csv")
async def export_users(request: Request):
    return StreamingResponse(
        csv_rows(request),
        media_type="text/csv",
        headers={
            "Content-Disposition": 'attachment; filename="users.csv"',
            "X-Accel-Buffering": "no",   # stop nginx buffering the whole body
        },
    )

Answer

Use the `@app.websocket('/path')` decorator. The handler receives a `WebSocket` object you can `accept()`, `receive_text()`/`receive_json()`, and `send_text()`/`send_json()` on. For broadcasting (chat, notifications), maintain a connection registry and iterate when sending.

For production scale, use Redis Pub/Sub between worker processes, because a `set` of sockets only knows about connections in its own process and with four Uvicorn workers three quarters of your users miss every message. Details interviewers look for. Authentication cannot use an `Authorization` header, since the browser `WebSocket` constructor cannot set one, so the token travels as a query parameter or a cookie, or as a first message immediately after connect; validate before `accept()` and call `await ws.close(code=4401)` to reject, since raising `HTTPException` in a WebSocket route does nothing useful.

Dependencies do work in WebSocket routes, including `Depends` with `yield`, but the teardown runs when the socket closes, so a database session held for a 30-minute chat connection will exhaust the pool: acquire per message instead. `WebSocketDisconnect` must be caught and the registry entry removed in a `finally`, otherwise you leak sockets and eventually broadcast into dead connections. Sending to a slow client blocks the coroutine, so real systems put a bounded `asyncio.Queue` per connection and drop or disconnect when it fills. Also send periodic pings, since idle connections get killed by ALBs and nginx after 60 seconds by default, and remember that scaling out needs sticky sessions or a shared pub/sub layer.

import asyncio
from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
rooms: dict[str, set[WebSocket]] = {}

async def ws_user(ws: WebSocket, token: str | None = None) -> str:
    user = verify_jwt(token) if token else None
    if not user:
        await ws.close(code=4401)   # policy violation, before accept()
        raise WebSocketDisconnect(4401)
    return user

@app.websocket("/ws/{room}")
async def chat(ws: WebSocket, room: str, user: str = Depends(ws_user)):
    await ws.accept()
    rooms.setdefault(room, set()).add(ws)
    try:
        while True:
            msg = await asyncio.wait_for(ws.receive_text(), timeout=60)
            await redis.publish(f"room:{room}", f"{user}: {msg}")  # fan out across workers
    except (WebSocketDisconnect, asyncio.TimeoutError):
        pass
    finally:
        rooms[room].discard(ws)     # always clean up

Answer

Path params live in the URL path (`/users/{id}`) and are always required. Query params are URL query-string key-value pairs (`?q=hello&page=2`), typically optional. Body params come from the request body and are Pydantic models.

FastAPI determines which is which by inspecting your function signature: if a parameter's name appears in the path, it's a path param; if it's a Pydantic model, it's the body; otherwise it's a query param. The precise rule for the remaining cases is worth stating: a scalar type (int, str, float, bool, UUID, date, Enum) that is not in the path defaults to a query parameter, and a complex type (a `BaseModel`, a `dict`, or a `list` of models) defaults to the body, which is why `tags: list[str]` unexpectedly becomes a body field until you annotate it with `Query()`. Explicit markers (`Path()`, `Query()`, `Body()`, `Header()`, `Cookie()`, `Form()`, `File()`) override the inference in either direction, so `Body(embed=True)` forces a single scalar to be read from the JSON body.

If a handler declares two Pydantic models, FastAPI stops sending the raw model as the body and instead expects an object keyed by parameter name, a change that silently breaks clients when someone adds a second model. Semantics matter beyond mechanics: query parameters belong in caches and server logs, so tokens, OTPs and PII must never travel there, and GET requests should not carry a body because proxies and some clients drop it. Path parameters identify a resource, query parameters filter or paginate a collection, and the body carries state you are creating or changing.

from typing import Annotated
from uuid import UUID
from fastapi import Body, FastAPI, Path, Query
from pydantic import BaseModel

app = FastAPI()

class OrderPatch(BaseModel):
    qty: int | None = None

@app.patch("/tenants/{tenant_id}/orders/{order_id}")
async def patch_order(
    tenant_id: Annotated[UUID, Path()],                 # path
    order_id: Annotated[int, Path(ge=1)],               # path
    dry_run: Annotated[bool, Query()] = False,          # query
    tags: Annotated[list[str], Query()] = [],           # WITHOUT Query() this becomes body
    patch: OrderPatch = Body(...),                      # body
    reason: Annotated[str, Body(embed=True)] = "",      # scalar forced into the body
):
    return {"tenant": tenant_id, "order": order_id, "dry_run": dry_run,
            "tags": tags, "patch": patch.model_dump(exclude_unset=True), "reason": reason}

# PATCH /tenants/<uuid>/orders/7?dry_run=true&tags=a&tags=b
# body: {"patch": {"qty": 3}, "reason": "stock adjustment"}

Answer

For single-field validation, use `Query()` with `ge`, `le`, `min_length`, `max_length` and `pattern` arguments (`pattern` replaced the removed `regex` keyword in the Pydantic v2 era). For multi-field rules such as 'either email or phone is required' or 'date_to must be after date_from', declare a Pydantic model as a query model instead of individual parameters: FastAPI 0.115 added support for `Annotated[Filters, Query()]`, and on older versions the equivalent is `Depends(Filters)` on a plain class or dataclass. Cross-field logic then lives in a `model_validator(mode="after")`, which runs once all fields are parsed and can compare them.

Points a senior interviewer will press on. Raise `ValueError` inside the validator, not `HTTPException`: Pydantic catches the `ValueError` and FastAPI turns it into a proper 422 with the field location, whereas an `HTTPException` raised during parsing escapes the validation envelope and produces an inconsistent error shape. Use `mode="before"` when you need to normalise raw input (trimming, splitting a comma-separated `ids=1,2,3` into a list) and `mode="after"` for rules across already-typed fields.

Set `model_config = ConfigDict(extra="forbid")` on the query model so a typo like `?limitt=10` fails loudly instead of silently paginating with the default. Reusable constrained types (`PositiveInt`, or `Annotated[int, Field(ge=1, le=100)]` aliased as `PageSize`) keep the rules in one place and still render correctly in the OpenAPI schema, which is what generated clients and Swagger UI read.

from datetime import date
from typing import Annotated
from fastapi import FastAPI, Query
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

app = FastAPI()

class SearchFilter(BaseModel):
    model_config = ConfigDict(extra="forbid")   # ?limitt=10 now 422s

    email: str | None = None
    phone: str | None = None
    date_from: date | None = None
    date_to: date | None = None
    limit: int = Field(20, ge=1, le=100)
    ids: list[int] = []

    @field_validator("ids", mode="before")
    @classmethod
    def split_csv(cls, v):
        return v.split(",") if isinstance(v, str) else v   # ?ids=1,2,3

    @model_validator(mode="after")
    def check_combo(self):
        if not self.email and not self.phone:
            raise ValueError("email or phone required")     # ValueError -> 422
        if self.date_from and self.date_to and self.date_from > self.date_to:
            raise ValueError("date_from must be on or before date_to")
        return self

@app.get("/search")
async def search(filters: Annotated[SearchFilter, Query()]):
    return await find(filters)

Answer

FastAPI 0.100 (July 2023) switched to Pydantic v2, whose core moved to Rust. The rename list is what you hit first: `.dict()` becomes `.model_dump()`, `.json()` becomes `.model_dump_json()`, `parse_obj` becomes `model_validate`, `parse_raw` becomes `model_validate_json`, `copy` becomes `model_copy`, `.schema()` becomes `.model_json_schema()`, the inner `class Config` becomes `model_config = ConfigDict(...)`, `@validator` becomes `@field_validator` (and now needs `@classmethod` under it), `@root_validator` becomes `@model_validator(mode='before'|'after')`, `orm_mode` becomes `from_attributes`, `allow_population_by_field_name` becomes `populate_by_name`, and the `regex` argument becomes `pattern`. Behavioural changes bite harder than renames. `Optional[str]` no longer implies a default of `None`, so `name: str | None` is a required field that accepts null and you must write `= None` to make it optional; this silently turns working endpoints into 422s.

Coercion got stricter in places: int to str no longer happens implicitly, and float to int only when lossless. Validators run in a different order, and errors are now typed with machine-readable `type` codes such as `string_too_short`, so anything parsing error messages by text breaks. Unknown fields are still ignored by default. Migration approach: run `bump-pydantic` for the mechanical rewrites, upgrade FastAPI and Pydantic together (0.100+ supports v2, and v1 support was dropped later), then rely on tests that assert 422 shapes to catch the required-versus-optional regressions. `pydantic.v1` is importable as a compatibility shim if a dependency still needs the old API.

# Pydantic v1
from pydantic import BaseModel, validator

class User(BaseModel):
    class Config:
        orm_mode = True
        allow_population_by_field_name = True

    name: str = None          # implicitly optional in v1

    @validator("name")
    def strip(cls, v): return v.strip()

# Pydantic v2
from pydantic import BaseModel, ConfigDict, field_validator

class User(BaseModel):
    model_config = ConfigDict(from_attributes=True, populate_by_name=True)

    name: str | None = None   # the `= None` is now MANDATORY to be optional

    @field_validator("name")
    @classmethod
    def strip(cls, v: str | None) -> str | None:
        return v.strip() if v else v

# u.dict()  -> u.model_dump()
# User.parse_obj(d) -> User.model_validate(d)
# pip install bump-pydantic && bump-pydantic app/

Answer

`BaseSettings` moved out of Pydantic core into the separate `pydantic-settings` package in v2, so the import is `from pydantic_settings import BaseSettings, SettingsConfigDict`. You declare a settings class with typed fields, and values are resolved from environment variables first, then a `.env` file, then the field defaults. Typing does real work here: `DEBUG=false` in the environment is the string `false`, and a `bool` annotation parses it correctly while a hand-rolled `os.getenv("DEBUG")` returns a truthy string.

Missing required settings raise a validation error at import time, which is exactly what you want, the container fails to start rather than 500ing on the first request that touches the value. Use `SecretStr` for passwords and API keys so an accidental `print(settings)` or a logged model dump shows `**********` instead of the secret, and call `.get_secret_value()` at the point of use. Nested config uses `env_nested_delimiter="__"` so `DB__HOST` populates `settings.db.host`.

Instantiate once and inject the object as a cached dependency with `@lru_cache`, which keeps it overridable in tests via `dependency_overrides` instead of being an unpatchable module global. Practical rules: never commit `.env`, ship a `.env.example`, load real secrets from AWS Secrets Manager, Infisical or Kubernetes secrets injected as environment variables, and set `extra="ignore"` because a container environment always has variables your model does not declare. `case_sensitive=False` is the default, so `database_url` matches `DATABASE_URL`.

from functools import lru_cache
from typing import Annotated
from fastapi import Depends, FastAPI
from pydantic import Field, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_nested_delimiter="__", extra="ignore"
    )

    environment: str = "local"
    debug: bool = False                      # DEBUG=false parses correctly
    database_url: str                        # required: no default, fails fast
    jwt_secret: SecretStr                    # repr is **********
    pool_size: int = Field(10, ge=1, le=50)

@lru_cache
def get_settings() -> Settings:
    return Settings()

SettingsDep = Annotated[Settings, Depends(get_settings)]
app = FastAPI()

@app.get("/config")
async def config(s: SettingsDep):
    return {"env": s.environment, "debug": s.debug}

# tests: app.dependency_overrides[get_settings] = lambda: Settings(database_url="sqlite://", jwt_secret="x")

Answer

`@app.on_event("startup")` and `@app.on_event("shutdown")` were deprecated in FastAPI 0.93 in favour of a single `lifespan` async context manager passed to the `FastAPI(...)` constructor. Everything before the `yield` runs once per worker process before the server accepts traffic, everything after runs during shutdown. The practical advantage is shared scope: with the old decorators you stashed the database engine or HTTP client in a module-level global or on `app.state` and hoped the ordering worked out, while a context manager holds both in local variables, guarantees teardown pairs with setup even if startup partially fails, and composes with `contextlib.AsyncExitStack` when you have several resources.

You can also yield a dict, which Starlette merges into `request.state`, giving typed access without globals. What belongs in there: creating the httpx `AsyncClient` and the SQLAlchemy engine, opening the Redis pool, warming a cache or loading an ML model, starting a background `asyncio.Task`. What does not: database migrations, which belong in a separate job because every worker would run them concurrently.

Behaviour to know: lifespan runs once per process, so four Uvicorn workers execute it four times, and `TestClient` only triggers it when used as a context manager. If startup raises, Uvicorn logs `Application startup failed` and exits rather than serving broken traffic. On shutdown, close clients and cancel your background tasks with a `try/except asyncio.CancelledError`, otherwise the process hangs until the orchestrator's kill timeout.

from contextlib import asynccontextmanager
import asyncio
import httpx
from fastapi import FastAPI, Request
from sqlalchemy.ext.asyncio import create_async_engine

@asynccontextmanager
async def lifespan(app: FastAPI):
    engine = create_async_engine(settings.database_url, pool_size=10)
    client = httpx.AsyncClient(timeout=5.0, limits=httpx.Limits(max_connections=100))
    refresher = asyncio.create_task(refresh_cache_loop())
    yield {"engine": engine, "http": client}      # merged into request.state
    refresher.cancel()
    try:
        await refresher
    except asyncio.CancelledError:
        pass
    await client.aclose()
    await engine.dispose()

app = FastAPI(lifespan=lifespan)

@app.get("/proxy")
async def proxy(request: Request):
    r = await request.state.http.get("https://api.example.com/rates")
    return r.json()

# deprecated since 0.93:
# @app.on_event("startup")
# async def startup(): ...

Answer

Two designs, and the interview is really about knowing when each breaks. Offset pagination (`?limit=20&offset=200`, or page numbers) is trivial to implement and lets a client jump to page 50, but Postgres still walks and discards every skipped row, so `OFFSET 100000` gets slow, and if a row is inserted between requests items shift and the user sees a duplicate or misses one. Keyset (cursor) pagination sorts by an indexed, unique, monotonic key and asks for rows after the last one seen, `WHERE (created_at, id) < (:ts, :id) ORDER BY created_at DESC, id DESC LIMIT 20`, which is index-only and constant time no matter how deep you go, and it is stable under concurrent inserts.

The cost is no random page access and a slightly more complex client. Include the tiebreaker column, otherwise rows sharing a timestamp are skipped or repeated. Implementation details in FastAPI: express the parameters as a reusable dependency so every list endpoint agrees on the same contract, cap `limit` with `Field(le=100)` so nobody requests a million rows, and encode the cursor as opaque base64 so clients cannot craft one and you can change the internal shape later.

Fetch `limit + 1` rows to know whether there is a next page without running a second `COUNT`, which is the expensive part of offset pagination on large tables. Return a generic `Page[T]` Pydantic model so the OpenAPI schema documents `items`, `next_cursor` and `has_more` properly. The `fastapi-pagination` library packages all of this if you want it off the shelf.

import base64, json
from typing import Annotated, Generic, TypeVar
from fastapi import Depends, FastAPI, Query
from pydantic import BaseModel
from sqlalchemy import select, tuple_

T = TypeVar("T")
app = FastAPI()

class Page(BaseModel, Generic[T]):
    items: list[T]
    next_cursor: str | None = None
    has_more: bool = False

class Cursor(BaseModel):
    limit: Annotated[int, Query(ge=1, le=100)] = 20
    after: str | None = None

CursorDep = Annotated[Cursor, Depends()]

@app.get("/orders")
async def list_orders(p: CursorDep, db: DB) -> Page[OrderOut]:
    stmt = select(Order).order_by(Order.created_at.desc(), Order.id.desc())
    if p.after:
        c = json.loads(base64.urlsafe_b64decode(p.after))
        stmt = stmt.where(tuple_(Order.created_at, Order.id) < (c["ts"], c["id"]))
    rows = (await db.execute(stmt.limit(p.limit + 1))).scalars().all()
    has_more = len(rows) > p.limit
    rows = rows[: p.limit]
    nxt = None
    if has_more:
        last = rows[-1]
        nxt = base64.urlsafe_b64encode(
            json.dumps({"ts": last.created_at.isoformat(), "id": last.id}).encode()
        ).decode()
    return Page[OrderOut](items=rows, next_cursor=nxt, has_more=has_more)

Answer

Alembic is the migration tool for SQLAlchemy and it needs a little wiring when your app uses an async driver. Initialise with `alembic init -t async migrations`, which generates an `env.py` that uses `async_engine_from_config` and runs the migration body inside `connection.run_sync(do_run_migrations)`, because Alembic's internals are synchronous and only the connection is async. Point `target_metadata` at your declarative `Base.metadata` and import every model module there, otherwise autogenerate sees an empty schema and cheerfully writes a migration that drops all your tables.

Read the URL from your settings object rather than hard-coding it in `alembic.ini`, so the same migrations run against local, staging and production. Then `alembic revision --autogenerate -m "add orders.status"` and `alembic upgrade head`. What interviewers want to hear next is the operational side.

Autogenerate is a draft, not an answer: it does not detect column renames (it emits a drop plus an add, which destroys data), it misses server defaults, constraint renames and most type changes, and it needs `compare_type=True` to notice varchar length changes, so every generated file gets read before it is committed. Migrations must run as a separate step, an init container, a Kubernetes Job or a deploy step, never at app startup where N workers race each other. For zero-downtime deploys, use expand-and-contract: add a nullable column, backfill in batches, deploy code that writes both, then drop the old column in a later release. On Postgres, guard against long `ACCESS EXCLUSIVE` locks with `SET lock_timeout` and create indexes concurrently outside a transaction.

# migrations/env.py (async template)
from alembic import context
from sqlalchemy.ext.asyncio import async_engine_from_config
from app.models import Base            # import ALL model modules or autogenerate drops tables
from app.config import get_settings

config = context.config
config.set_main_option("sqlalchemy.url", get_settings().database_url)
target_metadata = Base.metadata

def do_run_migrations(connection):
    context.configure(connection=connection, target_metadata=target_metadata,
                      compare_type=True, compare_server_default=True)
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    engine = async_engine_from_config(config.get_section(config.config_ini_section))
    async with engine.connect() as conn:
        await conn.run_sync(do_run_migrations)   # Alembic core is sync
    await engine.dispose()

# a safe online index build lives outside the transaction
# op.execute("COMMIT")
# op.create_index("ix_orders_status", "orders", ["status"], postgresql_concurrently=True)

# alembic revision --autogenerate -m "add orders.status"
# alembic upgrade head      # run as a job, never on app startup

Answer

The biggest wins (in order of impact): (1) Run multiple worker processes, either `uvicorn main:app --workers 4` or the Gunicorn manager `gunicorn main:app -k uvicorn.workers.UvicornWorker -w $(nproc)`; in containers, one worker per container with the orchestrator scaling replicas is now the more common shape because it gives per-process memory limits and clean rolling restarts. (2) Use async drivers everywhere, `asyncpg` for Postgres, `redis.asyncio` for Redis (the old `aioredis` package was merged into `redis-py` 4.2 and is no longer maintained separately). A single blocking call kills throughput. (3) Pool connections: database pool sized against your real database limit, a Redis pool, and one `httpx.AsyncClient` created in `lifespan` rather than per request, because building a new client per call re-does TLS handshakes and is a very common latency bug. (4) Cache aggressively, Redis cache-aside for hot reads plus a CDN in front of anything public. (5) Profile before guessing: `py-spy top --pid <pid>` on a live container needs no code change and shows exactly where CPU goes, usually Pydantic serialisation of large lists or JSON encoding; switch the route to `ORJSONResponse` for a 2-5x encoding speedup and skip `response_model` on very large payloads. Set `--http httptools` and `--loop uvloop` (the defaults when both are installed) and disable `--access-log` in high-traffic services since logging every request through Python costs real time. Measure with a load generator that reports p99 rather than average, and always verify the database is not the actual ceiling before tuning the app.

from contextlib import asynccontextmanager
import httpx
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(5.0, connect=2.0),
        limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
    )
    yield
    await app.state.http.aclose()

# app-wide default encoder
app = FastAPI(lifespan=lifespan, default_response_class=ORJSONResponse)

@app.get("/report", response_class=ORJSONResponse)
async def report():
    rows = await db.fetch_big_report()      # 50k rows
    return ORJSONResponse(rows)             # no response_model validation pass

# uvicorn main:app --workers 4 --loop uvloop --http httptools --no-access-log
# py-spy top --pid 1        # live flame view, no restart needed

Answer

Use `authlib` or `httpx-oauth` to handle the OAuth flow. The architecture: (1) An endpoint per provider that redirects to the provider's authorize URL with your client_id, redirect_uri and state, (2) A callback endpoint that exchanges the code for an access token, fetches the user profile, and either creates or looks up the local user, (3) Issue your own JWT to the client, don't return the provider's token. State must be cryptographically random and bound to the session to prevent CSRF.

Account linking is the trickiest part: if the same email logs in via Google AND GitHub, you need a strategy (link silently, prompt the user, or reject). Silent linking is only safe when the provider asserts a verified email, so check `email_verified` on the Google ID token and remember GitHub returns an unverified primary email unless you request the `user:email` scope and filter on `verified`. Use Authorization Code with PKCE even for a confidential server-side client, since it is now the baseline recommendation and mandatory for public clients; that means generating a `code_verifier`, sending its S256 `code_challenge` on the authorize call, and storing the verifier alongside the state.

Store `state` and `code_verifier` in a short-lived signed cookie or Redis key with a two to ten minute TTL rather than in a process dict, because a multi-worker deployment will land the callback on a different worker. Register the exact `redirect_uri` per environment; a mismatch is the single most common `redirect_uri_mismatch` failure. Persist provider identity as `(provider, provider_user_id)` rather than email, because emails change. Encrypt provider refresh tokens at rest if you keep them, and rely on the provider's discovery document (`/.well-known/openid-configuration`) plus cached JWKS to verify ID tokens rather than hard-coding endpoints.

import secrets
from authlib.integrations.starlette_client import OAuth
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import RedirectResponse
from starlette.middleware.sessions import SessionMiddleware

app = FastAPI()
app.add_middleware(SessionMiddleware, secret_key=settings.session_secret,
                   https_only=True, same_site="lax")

oauth = OAuth()
oauth.register(
    name="google",
    server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
    client_id=settings.google_id, client_secret=settings.google_secret,
    client_kwargs={"scope": "openid email profile", "code_challenge_method": "S256"},
)

@app.get("/auth/{provider}/login")
async def login(provider: str, request: Request):
    client = oauth.create_client(provider) or None
    if client is None:
        raise HTTPException(404, "unknown provider")
    redirect_uri = str(request.url_for("callback", provider=provider))
    return await client.authorize_redirect(request, redirect_uri, state=secrets.token_urlsafe(32))

@app.get("/auth/{provider}/callback", name="callback")
async def callback(provider: str, request: Request):
    token = await oauth.create_client(provider).authorize_access_token(request)  # verifies state
    info = token["userinfo"]
    if not info.get("email_verified"):
        raise HTTPException(400, "provider email not verified")
    user = await upsert_identity(provider, info["sub"], info["email"])
    return RedirectResponse(f"/app#token={make_jwt(user.id)}")

Answer

10k RPS on a single FastAPI service is achievable in 2026 but requires careful design. Stack: FastAPI with uvloop, asyncpg, Redis and multiple worker processes, deployed behind a load balancer (nginx, ALB) across several instances. Start from the arithmetic, because that is what a strong candidate does first: at 10k RPS and a 20 ms p50, Little's Law says roughly 200 requests are in flight at any moment, so the question becomes how many concurrent database round trips that implies and whether your connection budget covers it.

Database is almost always the bottleneck, use PgBouncer in transaction-pooling mode, read replicas for queries, and cache-aside in Redis for hot reads. Avoid synchronous CPU work in handlers, push image processing, video transcoding and ML inference to a worker queue (Celery, Dramatiq or SQS consumers). For predictable latency, add per-request timeouts and a circuit breaker on every downstream call, better to fail fast than to queue, because an unbounded queue converts a slow dependency into a total outage.

Add a concurrency limiter (a semaphore, or `--limit-concurrency` on Uvicorn) so an overload sheds load with 503 instead of growing latency without bound. Keep responses small, use keep-alive, and compress selectively. Observability: structured JSON logs with a request ID, RED metrics per route in Prometheus, and OpenTelemetry traces sampled at a few percent with errors always sampled. Then the honest caveat: 10k RPS of trivial JSON is easy, 10k RPS of transactional writes is a database and sharding problem, not a FastAPI problem.

import asyncio
from fastapi import FastAPI, HTTPException

app = FastAPI()

# bound concurrency to a slow downstream so it cannot swallow every worker slot
payments_gate = asyncio.Semaphore(50)

async def call_payments(payload: dict) -> dict:
    try:
        await asyncio.wait_for(payments_gate.acquire(), timeout=0.05)
    except asyncio.TimeoutError:
        raise HTTPException(503, "payments busy")     # shed load, do not queue
    try:
        r = await asyncio.wait_for(app.state.http.post(PSP_URL, json=payload), timeout=2.0)
    except asyncio.TimeoutError:
        raise HTTPException(504, "payments timeout")
    finally:
        payments_gate.release()
    return r.json()

@app.get("/products/{sku}")
async def product(sku: str):
    cached = await redis.get(f"p:{sku}")             # cache-aside on the hot read path
    if cached:
        return orjson.loads(cached)
    row = await db_replica.fetch_product(sku)        # reads go to a replica
    await redis.set(f"p:{sku}", orjson.dumps(row), ex=60)
    return row

# uvicorn main:app --workers 4 --limit-concurrency 500 --timeout-keep-alive 15
# pgbouncer: pool_mode = transaction, default_pool_size = 40

Answer

SSE is a one-way streaming protocol over plain HTTP, simpler than WebSockets for server-to-client push (notifications, live feeds, LLM streaming). Implementation: return a `StreamingResponse` with `media_type='text/event-stream'`, and yield `data: <json>\n\n` chunks. The client uses `EventSource`.

For multi-worker setups, you need a pub/sub layer (Redis Streams or Postgres `LISTEN/NOTIFY`) so each worker can push to its connected clients when an event fires. Don't forget `X-Accel-Buffering: no` if you're behind nginx, otherwise buffering breaks SSE. The wire format has more to it than the `data:` line: `event:` names a custom event type the client subscribes to with `addEventListener`, `id:` sets the last event ID that the browser automatically replays via the `Last-Event-ID` header on reconnect (which is how you make delivery resumable, so store the offset), `retry:` sets the reconnect delay in milliseconds, and a bare `: ping` comment line is the standard heartbeat that keeps proxies and load balancers from closing an idle connection.

Every message must end with a blank line or nothing is flushed. Compared with WebSockets, SSE is one-directional, runs over plain HTTP so it survives corporate proxies, reconnects automatically, and needs no separate upgrade handling; the costs are no binary frames, no browser support for custom headers on `EventSource` (so the token goes in a query string or cookie unless you use the `fetch`-based EventSource polyfill), and, on HTTP/1.1, the six-connections-per-origin browser limit, which HTTP/2 removes. Each open stream holds a worker coroutine, so cap concurrent streams, set `Cache-Control: no-cache` and `Connection: keep-alive`, disable GZip on the route, and check `await request.is_disconnected()` so an abandoned tab does not keep a Redis subscription alive forever.

import asyncio, json
from typing import Annotated
from fastapi import Depends, FastAPI, Header, Request
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_stream(request: Request, user_id: int, last_id: str | None):
    yield "retry: 3000\n\n"                      # client reconnect delay
    events = redis_stream(f"user:{user_id}", start=last_id or "$").__aiter__()
    while not await request.is_disconnected():
        try:
            e = await asyncio.wait_for(events.__anext__(), timeout=20)
        except asyncio.TimeoutError:
            yield ": ping\n\n"                   # comment frame keeps proxies open
            continue
        except StopAsyncIteration:
            break
        yield f"id: {e['id']}\nevent: {e['type']}\ndata: {json.dumps(e['payload'])}\n\n"

@app.get("/events")
async def events(
    request: Request,
    user_id: Annotated[int, Depends(current_user)],
    last_event_id: Annotated[str | None, Header()] = None,   # sent on reconnect
):
    return StreamingResponse(
        event_stream(request, user_id, last_event_id),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "Connection": "keep-alive",
                 "X-Accel-Buffering": "no"},
    )

Answer

Three patterns: (1) **Schema-per-tenant** (Postgres): one database, one schema per tenant. Switch via `SET search_path` in a dependency. Best isolation, harder to migrate because every release must run the migration across every schema. (2) **Row-level tenancy**: every table has a `tenant_id` column.

Inject `tenant_id` from the JWT into every query via a Repository pattern or SQLAlchemy event hook. Most common in SaaS, easier to scale, but a bug can leak tenant data, write a test that asserts ALL queries include `tenant_id`. (3) **Database-per-tenant**: complete isolation, expensive at scale (1000+ tenants becomes hard to operate). Pick (2) for SaaS under 10k tenants, (1) for regulated industries (finance, healthcare).

In all cases, use a FastAPI dependency to extract `tenant_id` from the auth token and propagate to the DB layer, never trust a request body field or a client-supplied header. The strongest version of the row-level answer does not rely on developers remembering the filter: turn on Postgres row-level security, add a policy comparing `tenant_id` to `current_setting('app.tenant_id')`, and have the session dependency issue `SET LOCAL app.tenant_id` at the start of every transaction so the database enforces isolation even when a query forgets its `WHERE`. Note that RLS is bypassed by superusers and table owners, so the app must connect as a non-owner role.

With a connection pool, always use `SET LOCAL` inside a transaction rather than `SET`, otherwise the tenant value leaks to the next request that reuses that connection, which is the same class of bug as caching a tenant-scoped object in a module-level dict. Other things to get right: include `tenant_id` in the leading position of composite indexes, key every cache entry by tenant, stamp it on log lines and traces, and add per-tenant rate limits so one customer's batch job cannot starve the rest.

from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

async def tenant_id(user: Annotated[Claims, Depends(current_user)]) -> str:
    return user.tenant_id            # from the signed token, never from the body

async def tenant_db(tid: Annotated[str, Depends(tenant_id)]):
    async with SessionLocal() as session:
        # SET LOCAL is transaction-scoped, so it cannot leak to the next
        # request that borrows this pooled connection
        # set_config(..., is_local=True) is the parameterizable equivalent of
        # SET LOCAL: Postgres utility statements like SET cannot take bind
        # parameters, and interpolating the tenant id into SQL would be an
        # injection hole in the isolation layer itself
        await session.execute(
            text("SELECT set_config('app.tenant_id', :tid, true)"), {"tid": tid}
        )
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

TenantDB = Annotated[AsyncSession, Depends(tenant_db)]

@app.get("/invoices")
async def invoices(db: TenantDB):
    # even without an explicit filter, RLS restricts the rows
    return (await db.execute(text("SELECT id, amount FROM invoices"))).mappings().all()

# ALTER TABLE invoices ENABLE ROW LEVEL SECURITY;
# CREATE POLICY tenant_isolation ON invoices USING
#   (tenant_id = current_setting('app.tenant_id')::uuid);

Answer

That symptom pattern (unrelated routes degrading together, health checks timing out, CPU well under saturation) is the signature of a coroutine that stops yielding to the loop. Confirm it before changing anything. `py-spy dump --pid 1 --locals` attaches to the running container without a restart and prints every thread's stack; a blocked loop shows the event-loop thread parked inside something that is not `epoll_wait`, typically `ssl.read`, `socket.recv` from `requests` or `psycopg2`, `time.sleep`, or a `bcrypt`/`hashlib` call. `py-spy top --pid 1` gives the same view live. For a reproducible case, run with `PYTHONASYNCIODEBUG=1` or `loop.set_debug(True)` and asyncio logs `Executing <Handle ...> took 0.512 seconds`, naming the callback.

A cheap always-on detector is a watchdog task that sleeps 0.25 seconds in a loop and logs whenever the measured wall time overshoots, since drift equals the time the loop was unavailable. Once located, the fix is one of three: swap the library for an async one (`httpx` for `requests`, `asyncpg` for `psycopg2`, `aioboto3` for `boto3`), push the call into a thread with `await run_in_threadpool(fn)` or `await asyncio.to_thread(fn)`, or make the route a plain `def` so FastAPI threadpools the whole thing. Add a guardrail so it does not recur: `asyncio.timeout` around outbound calls, explicit timeouts on every client (`httpx` defaults to none if you pass `timeout=None`), and a metric on event-loop lag scraped by Prometheus so the next occurrence pages you instead of being reported by users.

import asyncio, logging, time
from contextlib import asynccontextmanager
from fastapi import FastAPI

log = logging.getLogger("loop")

async def lag_watchdog(interval: float = 0.25, threshold: float = 0.1):
    while True:
        start = time.perf_counter()
        await asyncio.sleep(interval)
        lag = time.perf_counter() - start - interval
        EVENT_LOOP_LAG.set(lag)              # prometheus gauge
        if lag > threshold:
            log.warning("event loop blocked for %.3fs", lag)

@asynccontextmanager
async def lifespan(app: FastAPI):
    task = asyncio.create_task(lag_watchdog())
    yield
    task.cancel()

app = FastAPI(lifespan=lifespan)

# on the box:
#   pip install py-spy
#   py-spy dump --pid 1 --locals     # who is holding the loop right now
#   py-spy top  --pid 1              # live sampling profile
#   PYTHONASYNCIODEBUG=1 uvicorn main:app   # logs slow callbacks by name

Answer

FastAPI builds a dependency graph per request and memoises each node in a per-request cache keyed by the `(callable, security_scopes)` pair. So if `get_current_user` is required by three sub-dependencies and the route itself, it executes once and everyone receives the same object. This is what makes deep dependency trees affordable: the token is decoded once, the database session is created once, and the identity object is shared, which also means all four callers see the same instance and mutating it in one place is visible everywhere.

The cache lives only for that request, there is no application-level singleton, and it is cleared as the request finishes. `Depends(fn, use_cache=False)` opts a node out, so it runs once per place it appears. You want that when the dependency is intentionally non-idempotent or must produce a distinct value per use: generating a fresh idempotency key, allocating a separate database session for a side transaction that must survive a rollback of the main one, taking a new timestamp, or a factory that returns a per-consumer resource. Two subtleties that come up in senior interviews.

Caching is by callable identity, so `Depends(paginate)` used twice is shared, while two different partials or two different lambda wrappers are separate nodes even if the body is identical; that is also why parameterised dependencies are written as a factory returning a fresh inner function. And for application-level caching (settings, a JWKS document, a machine-learning model) the per-request cache is the wrong tool: wrap the provider in `functools.lru_cache` or build it once in `lifespan`, which keeps it overridable in tests through `dependency_overrides` rather than as a frozen module global.

import uuid
from functools import lru_cache
from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()
calls = {"user": 0, "reqid": 0}

async def get_current_user() -> str:
    calls["user"] += 1          # runs ONCE per request, however many depend on it
    return "u_1"

async def new_request_id() -> str:
    calls["reqid"] += 1         # runs once PER USE
    return str(uuid.uuid4())

async def audit(user: Annotated[str, Depends(get_current_user)]):
    return f"audit:{user}"

@lru_cache                      # application-level, survives across requests
def get_settings():
    return Settings()

@app.get("/demo")
async def demo(
    user: Annotated[str, Depends(get_current_user)],
    a: Annotated[str, Depends(audit)],
    id1: Annotated[str, Depends(new_request_id, use_cache=False)],
    id2: Annotated[str, Depends(new_request_id, use_cache=False)],
    settings: Annotated[Settings, Depends(get_settings)],
):
    # calls == {"user": 1, "reqid": 2}; id1 != id2
    return {"user": user, "audit": a, "ids": [id1, id2], "env": settings.environment}

Answer

`TestClient` runs the app through a synchronous bridge, which is fine until your test itself needs to await something: seeding data through the async session, asserting on a Redis key, or driving a WebSocket alongside HTTP calls. Then you want `httpx.AsyncClient` talking to the app in-process. The wiring changed: `AsyncClient(app=app)` was deprecated and removed in httpx 0.28, so the current form is `AsyncClient(transport=ASGITransport(app=app), base_url="http://test")`.

Tests that used the old keyword fail with an unexpected-argument `TypeError`, and it is a common upgrade break. `ASGITransport` does not run the lifespan, so anything created in your `lifespan` context manager will be missing; either wrap the app with `LifespanManager` from `asgi-lifespan`, or build those resources in fixtures. You also need an async test runner, `pytest-asyncio` with `asyncio_mode = auto` in `pyproject.toml` (or `anyio_mode`), otherwise coroutine tests are collected and skipped with a warning rather than failing loudly. The database pattern that scales: a session-scoped engine, a function-scoped fixture that opens a connection, begins an outer transaction, binds the session to it, overrides `get_db`, and rolls back at the end, so tests are isolated without recreating the schema each time.

Keep the event loop consistent, a session-scoped engine plus a function-scoped loop is the classic `attached to a different loop` error. Stub outbound HTTP with `respx`, which patches at the httpx transport layer, and reserve `TestClient` for the cases where its `with` block is genuinely simpler.

# pyproject.toml: [tool.pytest.ini_options] asyncio_mode = "auto"
import pytest
from asgi_lifespan import LifespanManager
from httpx import ASGITransport, AsyncClient
from main import app, get_db

@pytest.fixture
async def session():
    conn = await engine.connect()
    trans = await conn.begin()
    s = AsyncSession(bind=conn, expire_on_commit=False)
    yield s
    await s.close()
    await trans.rollback()          # every test starts from a clean slate
    await conn.close()

@pytest.fixture
async def client(session):
    app.dependency_overrides[get_db] = lambda: session
    async with LifespanManager(app):                       # runs lifespan startup
        transport = ASGITransport(app=app)                 # httpx 0.28+
        async with AsyncClient(transport=transport, base_url="http://test") as c:
            yield c
    app.dependency_overrides.clear()

async def test_create_and_read(client, session):
    r = await client.post("/users", json={"email": "a@b.co", "age": 30, "full_name": "A"})
    assert r.status_code == 201
    got = await client.get(f"/users/{r.json()['id']}")
    assert got.json()["email"] == "a@b.co"

Answer

Threads do not help CPU-bound Python on CPython 3.13 and earlier, because the GIL serialises bytecode execution, so `run_in_threadpool` around a pure-Python hot loop just moves the stall. Decide by the shape of the work. Short bursts of a few milliseconds (password hashing with argon2, a small image thumbnail, a numpy or pandas operation) can go to a thread, and that genuinely works for libraries that release the GIL in C: numpy, pillow, bcrypt, cryptography, and pandas for many operations.

Anything longer than roughly 50 to 100 ms belongs off the request path entirely. Use a `ProcessPoolExecutor` created once in `lifespan` and driven with `loop.run_in_executor`, which sidesteps the GIL at the cost of pickling arguments and results, so it suits compute-heavy calls with small payloads. Never create the pool per request, forking on every call is slower than the work itself, and never fork after the event loop starts on Linux without setting the `spawn` start method, since forking a process that holds asyncio and database file descriptors leads to hangs.

For anything user-visible and slow (video transcoding, PDF generation, bulk report building, large ML inference) the correct architecture is a job queue plus a status endpoint: accept the request, return 202 with a job ID, run the work in Celery, Dramatiq or a dedicated inference service, and let the client poll or subscribe to SSE. Also cap the AnyIO threadpool deliberately, since its 40-thread default means 41 concurrent sync handlers queue. Python 3.13's free-threaded build changes this calculus but is not yet the default for production wheels in 2026.

import asyncio
from concurrent.futures import ProcessPoolExecutor
from contextlib import asynccontextmanager
from typing import Annotated
import anyio.to_thread
from fastapi import Body, FastAPI

def fingerprint(data: bytes) -> str:
    return heavy_pure_python_hash(data)     # holds the GIL, needs a process

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.pool = ProcessPoolExecutor(max_workers=4)   # created ONCE
    anyio.to_thread.current_default_thread_limiter().total_tokens = 80
    yield
    app.state.pool.shutdown(wait=True)

app = FastAPI(lifespan=lifespan)

@app.post("/fingerprint")
async def make_fingerprint(payload: Annotated[bytes, Body(media_type="application/octet-stream")]):
    loop = asyncio.get_running_loop()
    return {"hash": await loop.run_in_executor(app.state.pool, fingerprint, payload)}

@app.post("/login")
async def login(password: str):
    # bcrypt/argon2 release the GIL in C, a thread is enough
    ok = await anyio.to_thread.run_sync(verify_password, password)
    return {"ok": ok}

@app.post("/reports", status_code=202)
async def build_report(spec: dict):
    job_id = await queue.enqueue("build_report", spec)   # minutes of work: off-box
    return {"job_id": job_id, "status_url": f"/reports/{job_id}"}

Answer

Clients retry. A mobile app on a flaky connection resends after a timeout, a load balancer retries a 502, and a webhook provider redelivers until it sees a 2xx, so any endpoint that creates or charges must tolerate duplicates. The standard mechanism is an `Idempotency-Key` header: the client generates a UUID per logical operation and reuses it on every retry of that operation.

The server atomically claims the key, does the work, stores the response, and replays the stored response for any later request presenting the same key. Atomicity is the whole trick, so use `SET key value NX EX 86400` in Redis or an `INSERT ... ON CONFLICT DO NOTHING` against a unique index; a read-then-write check races under concurrency and charges twice.

Handle the in-flight case explicitly: if the key exists but has no stored response yet, the first request is still running, and the honest answer is 409 with a `Retry-After` rather than starting a second charge. Bind the key to the request payload by hashing the body, so a client reusing a key with different data gets a 422 instead of silently receiving someone else's result, and scope the key per user or tenant so keys cannot collide or be guessed across accounts. Store the status code and body for 24 hours, which is what Stripe and Razorpay do.

For internal state changes prefer natural idempotency where you can get it: a unique constraint on `(order_id, provider_ref)`, an upsert, or a state machine that treats a transition to its current state as a no-op. Webhook consumers get this for free by keying on the provider's event ID.

import hashlib, json
from dataclasses import dataclass
from typing import Annotated
from fastapi import Depends, FastAPI, Header, HTTPException, Request

app = FastAPI()
TTL = 86400

@dataclass
class Slot:
    key: str
    digest: str
    replay: dict | None      # not None => already completed, just return it

async def idempotency_slot(
    request: Request,
    user: Annotated[str, Depends(current_user)],
    idempotency_key: Annotated[str | None, Header()] = None,
) -> Slot:
    if not idempotency_key:
        raise HTTPException(400, "Idempotency-Key header required")
    digest = hashlib.sha256(await request.body()).hexdigest()
    key = f"idem:{user}:{idempotency_key}"
    # atomic claim: SET NX is the whole concurrency guarantee
    if await redis.set(key, json.dumps({"digest": digest}), nx=True, ex=TTL):
        return Slot(key, digest, None)
    stored = json.loads(await redis.get(key))
    if stored["digest"] != digest:
        raise HTTPException(422, "key reused with a different body")
    if "response" not in stored:
        raise HTTPException(409, "in progress", headers={"Retry-After": "1"})
    return Slot(key, digest, stored["response"])

@app.post("/payments")
async def charge(amount: int, slot: Annotated[Slot, Depends(idempotency_slot)]):
    if slot.replay is not None:
        return slot.replay                  # retry: same answer, no second charge
    r = await psp.charge(amount)
    body = {"id": r.id, "status": r.status}
    await redis.set(slot.key, json.dumps({"digest": slot.digest, "response": body}), ex=TTL)
    return body

Answer

A rolling deploy that returns 502s to users is almost always a shutdown-sequencing problem, not a FastAPI bug. When a pod is deleted, two things happen concurrently: the kubelet sends SIGTERM, and the endpoints controller starts removing the pod from Service endpoints, which then has to propagate to kube-proxy and to every ingress or load balancer. The propagation is slower, so for a second or two after SIGTERM the load balancer is still sending new requests to a process that has already stopped accepting them.

The fix is a `preStop` hook that sleeps 5 to 15 seconds before the signal is delivered to the app, giving deregistration time to land, plus a readiness probe you can flip to failing so the pod is pulled from rotation deliberately. On the application side, Uvicorn handles SIGTERM by refusing new connections and waiting for in-flight requests to finish, then runs the shutdown half of `lifespan`, where you close the httpx client, dispose the SQLAlchemy engine, cancel background tasks and flush the OpenTelemetry exporter. Make sure the signal actually reaches Python: an entrypoint written as a shell string makes the shell PID 1 and swallows SIGTERM, so use the exec form (`CMD ["uvicorn", ...]`) or `tini`. `terminationGracePeriodSeconds` must exceed preStop sleep plus your longest request, otherwise SIGKILL truncates it, and `--timeout-graceful-shutdown` on Uvicorn caps the wait.

Long-lived connections need explicit handling: SSE and WebSocket streams do not drain on their own, so send a close frame or a terminal event and let clients reconnect to a healthy pod. Separate `/livez` from `/readyz` so a slow dependency does not get your pod restarted.

from contextlib import asynccontextmanager
from fastapi import FastAPI

STATE = {"ready": True}

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient()
    yield                       # ---- SIGTERM lands here ----
    STATE["ready"] = False      # readiness starts failing immediately
    await app.state.http.aclose()
    await engine.dispose()
    tracer_provider.shutdown()  # flush spans before the process exits

app = FastAPI(lifespan=lifespan)

@app.get("/livez", include_in_schema=False)
async def livez():
    return {"status": "ok"}          # process is alive; never checks dependencies

@app.get("/readyz", include_in_schema=False)
async def readyz():
    if not STATE["ready"]:
        return JSONResponse({"status": "draining"}, status_code=503)
    return {"status": "ok"}

# Dockerfile: exec form so uvicorn is PID 1 and receives SIGTERM
# CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--timeout-graceful-shutdown", "25"]
#
# deployment.yaml
#   terminationGracePeriodSeconds: 45
#   lifecycle:
#     preStop:
#       exec: { command: ["sleep", "10"] }   # let endpoint removal propagate

Answer

Install `opentelemetry-instrumentation-fastapi` and call `FastAPIInstrumentor.instrument_app(app)`, which wraps the ASGI layer and creates a server span per request carrying `http.method`, `http.route` (the template, not the expanded path, which is what keeps cardinality sane), `http.status_code` and the trace context extracted from the incoming `traceparent` header. Add the SQLAlchemy, httpx or requests, and Redis instrumentors so downstream calls become child spans, which is how you see that a 900 ms endpoint is really 40 sequential queries. Export over OTLP to a collector rather than directly to a vendor, so switching backends is a config change.

Correlation is the part people forget: pull the active span context in a logging filter and add `trace_id` and `span_id` to every structured log line, then a slow trace links straight to its logs. Use `contextvars` for request-scoped fields such as request ID, user ID and tenant ID, because a plain global is shared across concurrently running coroutines in the same worker and will attribute one user's log line to another. Practical guidance: exclude health and metrics endpoints with `OTEL_PYTHON_FASTAPI_EXCLUDED_URLS=healthz,readyz,metrics` so they do not dominate your span volume and bill, use parent-based ratio sampling at a low percentage with a tail or error-based rule so failures are always kept, and never put tokens, request bodies or PII into span attributes. Add domain attributes that make traces searchable (`tenant.id`, `order.id`), keep RED metrics in Prometheus for alerting and use traces for diagnosis, and remember to flush the tracer provider on shutdown or the last spans of a terminating pod are lost.

import logging
from fastapi import FastAPI
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor

app = FastAPI()
FastAPIInstrumentor.instrument_app(app, excluded_urls="healthz,readyz,metrics")
HTTPXClientInstrumentor().instrument()
SQLAlchemyInstrumentor().instrument(engine=engine.sync_engine)

class TraceFilter(logging.Filter):
    def filter(self, record):
        ctx = trace.get_current_span().get_span_context()
        record.trace_id = format(ctx.trace_id, "032x") if ctx.is_valid else "-"
        record.span_id = format(ctx.span_id, "016x") if ctx.is_valid else "-"
        return True

logging.getLogger().addFilter(TraceFilter())
# format: '%(asctime)s %(levelname)s trace_id=%(trace_id)s span_id=%(span_id)s %(message)s'

@app.get("/orders/{order_id}")
async def get_order(order_id: int):
    span = trace.get_current_span()
    span.set_attribute("order.id", order_id)   # searchable; no PII, no tokens
    return await service.load(order_id)

# OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4317
# OTEL_TRACES_SAMPLER=parentbased_traceidratio OTEL_TRACES_SAMPLER_ARG=0.05

Answer

Measure the split first, because serialisation cost has two distinct halves and they have different fixes. Half one is `response_model` validation: FastAPI runs your return value through the model a second time, on top of whatever the ORM already built, so 50,000 rows means 50,000 model constructions. Half two is JSON encoding, where the stdlib `json` module is the slow part. `py-spy` will tell you which dominates, typically it is the model pass.

Fixes in order of payoff. Drop `response_model` on genuinely large collection endpoints and return an `ORJSONResponse` directly, which skips the revalidation entirely, at the cost of losing automatic field filtering, so only do it where the query already selects exactly the public columns. Switch the encoder: `ORJSONResponse` (orjson, Rust) is roughly 2 to 5 times faster than the default `JSONResponse` and handles `datetime`, `UUID` and `dataclass` natively, and you can make it the default with `FastAPI(default_response_class=ORJSONResponse)`.

Avoid `jsonable_encoder` in hot paths, it walks the structure in pure Python. Select only the columns you return rather than hydrating full ORM objects with relationships, since `selectinload` on a 50k-row query is usually the real cost hiding behind a serialisation complaint. Then question the payload itself: paginate, let the client request sparse fieldsets, cache the rendered bytes in Redis when the data is shared, and enable `GZipMiddleware(minimum_size=1000)` because network transfer often exceeds encoding time. If the response is genuinely huge and used for export, stream NDJSON or CSV instead so memory stays flat and the client can start processing immediately.

from fastapi import FastAPI
from fastapi.responses import ORJSONResponse, StreamingResponse
from sqlalchemy import select
import orjson

app = FastAPI(default_response_class=ORJSONResponse)

# SLOW: full ORM hydration + a second Pydantic validation pass
@app.get("/reports/slow", response_model=list[RowOut])
async def slow(db: DB):
    return (await db.execute(select(Row))).scalars().all()

# FAST: select only what you return, skip the model round trip
@app.get("/reports/fast", response_class=ORJSONResponse)
async def fast(db: DB):
    rows = (await db.execute(select(Row.id, Row.amount, Row.created_at))).mappings().all()
    return ORJSONResponse([dict(r) for r in rows])

# FLAT MEMORY: stream NDJSON for exports
@app.get("/reports/export")
async def export(db: DB):
    async def gen():
        async with engine.connect() as conn:
            async for r in await conn.stream(select(Row.id, Row.amount)):
                yield orjson.dumps({"id": r.id, "amount": r.amount}) + b"\n"
    return StreamingResponse(gen(), media_type="application/x-ndjson")