1.3 — FastAPI

1. What is it?

FastAPI is a Python web framework built on top of Starlette (ASGI server toolkit) and Pydantic (validation). It is the dominant framework for serving AI backends in Python because it's async-native and generates request/response validation and OpenAPI docs directly from Python type hints.

2. Why does it exist?

Before FastAPI, Python web frameworks were largely WSGI-based (Flask, Django) — synchronous by design, bolting on async support awkwardly. AI backends are fundamentally I/O-bound (waiting on LLM calls, vector DB queries), so a framework built async-first from day one, on top of ASGI, matches the workload far better. FastAPI also solved a second, less obvious problem: keeping your validation schema, your documentation, and your actual code from drifting apart — because all three are generated from the same Pydantic models and type hints.

3. What problem does it solve?

It solves "how do I serve concurrent AI requests correctly, with input validated at the boundary, and with documentation that doesn't lie" — all three at once, with minimal boilerplate.

4. How does it work internally?

ASGI and the request lifecycle

FastAPI runs on ASGI (Asynchronous Server Gateway Interface), served typically by Uvicorn. Each incoming HTTP connection is handled as an async task on the event loop (see 1.1). FastAPI's routing layer inspects your function signature's type hints at startup, builds a Pydantic validator for the request body/query/path params, and at request time: parses the raw ASGI scope/receive/send into a request, validates it against your declared types, calls your handler (awaiting it if it's async def), serializes the return value against your declared response_model, and sends the response.

UvicornASGI server · parses raw HTTP

Starletterouting, middleware chain

FastAPIDI, Pydantic validation, OpenAPI

Your route handlerasync def or def

Every request falls through this stack before your code runs.

Dependency Injection (Depends)

FastAPI's Depends() system resolves a function's own dependencies before calling it — used constantly for auth, DB sessions, and shared clients:

async def get_db_session() -> AsyncSession:
    async with SessionLocal() as session:
        yield session

@app.get("/conversations/{id}")
async def get_conversation(id: str, db: AsyncSession = Depends(get_db_session)):
    ...

The yield-based dependency pattern gives you setup/teardown (open session → run handler → close session, even on exception) for free — this is how DB connections, LLM clients, and per-request tracing spans are typically scoped.

Sync vs async route handlers

A def (not async def) route handler is automatically run in FastAPI's thread pool, not on the event loop — this is a deliberate escape hatch for calling blocking code without freezing the loop, but it means sync and async handlers have genuinely different concurrency behavior and resource costs (thread pool has a fixed size; exhausting it queues requests).

BackgroundTasks — fire-and-forget work after the response

BackgroundTasks runs a function after the response has already been sent to the client — for work the caller doesn't need to wait on (logging an analytics event, kicking off a re-embedding job, sending a notification):

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def log_query_for_analytics(query_text: str, user_id: str) -> None:
    """Writes a query event to the analytics store, off the request's critical path."""
    analytics_client.record(event="query", text=query_text, user_id=user_id)


@app.post("/v1/ask")
async def ask(query_text: str, user_id: str, background_tasks: BackgroundTasks) -> dict:
    """
    Answers a query immediately and schedules analytics logging to run after the
    response is returned, so the client isn't kept waiting on non-essential work.
    """
    answer = await run_llm_query(query_text)
    background_tasks.add_task(log_query_for_analytics, query_text, user_id)
    return {"answer": answer}

BackgroundTasks still runs inside the same process and event loop, after the response is sent — it's the right tool for small, fast, best-effort work. It is not a durable job queue: if the process crashes before a background task runs, the task is lost, and a slow background task still occupies the worker. Genuinely important or slow async work (bulk re-embedding a document corpus, sending a webhook with retries) belongs in Celery/RQ (Part 7), not BackgroundTasks.

File uploads (UploadFile) for document ingestion

Document-intelligence and RAG-ingestion endpoints — "upload this PDF/contract/invoice and we'll process it" — are one of the most common FDE-built FastAPI routes. UploadFile streams the file to a temporary spooled file rather than loading it entirely into memory, which matters once uploads are multi-megabyte PDFs rather than tiny JSON bodies:

from fastapi import FastAPI, UploadFile, HTTPException

app = FastAPI()

ALLOWED_CONTENT_TYPES = {"application/pdf"}
MAX_UPLOAD_BYTES = 25 * 1024 * 1024  # 25 MB


@app.post("/v1/documents")
async def ingest_document(file: UploadFile) -> dict:
    """
    Accepts an uploaded document for RAG ingestion, validating type and size
    before any parsing or embedding work is attempted.

    Args:
        file (UploadFile): The uploaded file, streamed rather than fully buffered.

    Returns:
        dict: The created document's id and chunk count.
    """
    if file.content_type not in ALLOWED_CONTENT_TYPES:
        raise HTTPException(status_code=415, detail=f"unsupported content type: {file.content_type}")

    contents = await file.read(MAX_UPLOAD_BYTES + 1)
    if len(contents) > MAX_UPLOAD_BYTES:
        raise HTTPException(status_code=413, detail="file exceeds maximum upload size")

    document_id = await ingest_and_chunk_pdf(contents, filename=file.filename)
    return {"document_id": document_id}

Validate content_type and enforce a size cap before handing the bytes to a PDF parser or an embedding pipeline — an unbounded upload accepted straight into a parsing library is both a resource-exhaustion risk and, per Part 9, an untrusted-input boundary exactly like LLM output.

Testing with TestClient and dependency_overrides

Section 19 claims Depends() matters for testability — here's the payoff: app.dependency_overrides lets a test swap a real dependency (a DB session, an LLM client) for a test double, without changing a single line of route code:

from fastapi.testclient import TestClient
from myapp.main import app
from myapp.dependencies import get_llm_client

app.dependency_overrides[get_llm_client] = lambda: FakeLLMClient(canned_response="mocked summary")

client = TestClient(app)


def test_summarize_returns_llm_output():
    """The route returns whatever the injected (fake) LLM client produces."""
    response = client.post("/v1/summaries", json={"text": "some long document"})
    assert response.status_code == 200
    assert response.json()["summary"] == "mocked summary"

Because the route handler only ever depends on the interface Depends(get_llm_client) returns, the test never touches the real Anthropic API, never costs money, and never flakes on network variance — this is exactly the testability payoff dependency injection is supposed to buy, made concrete instead of asserted. dependency_overrides is a dict on the app object, so remember to clear it (app.dependency_overrides.clear()) between tests that need different fakes, typically via a pytest fixture. Part 1.9 shows the equivalent pattern for an async def route using httpx.ASGITransport.

5. Simple mental model

FastAPI is a contract-enforcing receptionist: it reads the type hints you wrote as the "form" a request must fill out, rejects anything that doesn't match the form before your business logic ever runs, and automatically publishes the form as documentation (/docs) so anyone integrating with you can see the contract without asking you.

6. Real-world example

A logistics company's AI backend exposes /v1/shipments/{id}/eta-explanation — an LLM-generated natural-language explanation of a shipment's ETA. FastAPI validates the path parameter is a valid shipment ID format, injects a DB session dependency to fetch the shipment, injects an authenticated-user dependency to check RBAC (can this caller see this shipment), and only then calls the LLM. If any of those steps fail, the LLM is never invoked — saving cost and avoiding leaking data to unauthorized callers.

7. Architecture diagram

Client

Uvicorn

FastAPI routing + validation

Auth / DB / LLM client depsresolved via Depends()

Route handlerbusiness logic + LLM/RAG calls

8. Production considerations

• Construct expensive shared resources (DB connection pool, LLM client) once at startup via FastAPI's lifespan context manager, not per-request.
• Use response_model to control exactly what's serialized back — prevents accidentally leaking internal fields (e.g., raw LLM reasoning, internal IDs) that happen to exist on your internal data model.
• Run behind a process manager (Uvicorn workers, or Gunicorn with Uvicorn workers) for multi-core utilization — one Uvicorn process is still bound by one event loop/GIL for CPU-bound work.
• Add middleware for request ID propagation and structured logging — essential for debugging a customer's specific failed request in production (Part 8).

9. Common mistakes

• Defining a route as def (sync) by habit, then doing await inside it — this is a syntax error, but the more insidious version is defining it async def and then calling a blocking library inside, silently stalling the loop (see 1.1).
• Not using response_model, so an internal field (an API key, an internal reasoning trace, another tenant's data accidentally joined in) leaks into the response.
• Rebuilding a DB connection or LLM client inside every request instead of injecting a shared instance — kills performance and can exhaust connection limits.

10. Security considerations

• FastAPI validates shape, not authorization — a well-typed request can still be for a resource the caller shouldn't access. Authorization must be an explicit dependency (Part 9), not assumed from validation passing.
• CORS misconfiguration (allow_origins=["*"] with credentials) is a common FastAPI mistake that exposes an AI API to cross-origin abuse.
• Don't return raw exception messages/stack traces to clients in production (debug=True should never be enabled in production) — internal error details can leak infrastructure information.

11. Performance considerations

• Prefer async def handlers with async dependencies wherever possible; use the sync-in-threadpool escape hatch only for unavoidable blocking calls, and be aware of the thread pool's fixed size as a throughput ceiling.
• Streaming responses (StreamingResponse) for token-by-token LLM output avoid buffering the entire response in memory and reduce perceived latency.

12. Cost considerations

• A poorly bounded endpoint (no auth, no rate limit) that triggers LLM calls is a direct line from a public API to your LLM bill — FastAPI doesn't rate-limit by default; add it explicitly (middleware or an API gateway).

13. When to use it

Any Python-served HTTP API for an AI system, especially one that's I/O-bound and benefits from async — which is essentially the default choice today for new AI backends in Python.

14. When NOT to use it

• If your team is standardized on Django and you need its ORM/admin/ecosystem more than async performance, forcing FastAPI in adds friction for little gain.
• Extremely simple internal scripts don't need a full ASGI web framework.

15. Alternatives and trade-offs

Framework	Good for	Weak point
FastAPI	Async-native APIs, auto docs, type-driven validation	Less batteries-included than Django (no built-in admin/ORM)
Flask	Simplicity, huge ecosystem, familiarity	Sync by default; async support is bolted on
Django (+ DRF)	Full-featured apps needing ORM/admin/auth out of the box	Heavier; async support newer/less idiomatic

16. Practical Python/code example

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from httpx import AsyncClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Creates a single shared HTTP client for the app's lifetime and closes it on shutdown."""
    app.state.llm_client = AsyncClient(base_url="https://api.anthropic.com", timeout=30.0)
    yield
    await app.state.llm_client.aclose()


app = FastAPI(lifespan=lifespan)


def get_llm_client() -> AsyncClient:
    """Provides the shared LLM HTTP client to route handlers via dependency injection."""
    return app.state.llm_client


@app.post("/v1/summaries")
async def summarize(text: str, client: AsyncClient = Depends(get_llm_client)) -> dict:
    """
    Summarizes the given text using the shared LLM client.

    Args:
        text (str): Text to summarize.
        client (AsyncClient): Shared, connection-pooled HTTP client for the LLM provider.

    Returns:
        dict: The summary result.
    """
    response = await client.post("/v1/messages", json={"input": text})
    response.raise_for_status()
    return {"summary": response.json()["output"]}

17. Production-quality example

import logging
import uuid
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request, Depends, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware

logger = logging.getLogger("ai_backend")


class RequestIDMiddleware(BaseHTTPMiddleware):
    """Attaches a unique request ID to every request for log correlation."""

    async def dispatch(self, request: Request, call_next):
        request_id = str(uuid.uuid4())
        request.state.request_id = request_id
        response = await call_next(request)
        response.headers["X-Request-ID"] = request_id
        return response


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.llm_client = build_llm_client()
    app.state.db_pool = await build_db_pool()
    yield
    await app.state.llm_client.aclose()
    await app.state.db_pool.close()


app = FastAPI(lifespan=lifespan)
app.add_middleware(RequestIDMiddleware)


async def get_current_user(request: Request, authorization: str | None = None) -> "User":
    """
    Resolves and authorizes the caller from the request's bearer token.

    Args:
        request (Request): The incoming request, used to log the resolved identity.
        authorization (str | None): The Authorization header value.

    Returns:
        User: The authenticated user.
    """
    if authorization is None:
        raise HTTPException(status_code=401, detail="missing Authorization header")
    user = await resolve_user_from_token(authorization)
    logger.info("request_id=%s user=%s", request.state.request_id, user.id)
    return user


@app.post("/v1/summaries")
async def summarize(
    text: str,
    request: Request,
    user: "User" = Depends(get_current_user),
) -> dict:
    """
    Summarizes text on behalf of an authenticated user, scoped to their tenant.

    Args:
        text (str): Text to summarize.
        request (Request): Incoming request, used for correlation logging.
        user (User): The authenticated, authorized caller.

    Returns:
        dict: The summary result.
    """
    if len(text) > 50_000:
        raise HTTPException(status_code=422, detail="text exceeds maximum length")
    summary = await run_summary_for_tenant(text, tenant_id=user.tenant_id)
    return {"summary": summary, "request_id": request.state.request_id}

18. Short exercise

Add a dependency to the production example that enforces a per-user rate limit (e.g., 10 requests/minute) using an in-memory counter, and returns 429 with a Retry-After header when exceeded. Then consider: why would this in-memory approach break in a multi-instance deployment, and what would you replace it with (see Part 7 caching/Redis)?

19. Interview questions

• Explain what happens differently when a route handler is def vs async def in FastAPI.
• Why does FastAPI's Depends() system matter for testability of an AI backend? Show how dependency_overrides cashes that claim in a real test.
• How would you prevent a customer's PII from accidentally appearing in a FastAPI response body?
• When would you use BackgroundTasks versus a real job queue (Celery/RQ) for post-response work?
• What validation should happen before an uploaded document ever reaches a parsing or embedding pipeline, and why?

20. FDE/customer scenario

Customer: "Can you show me the API docs for the assistant endpoint before we build our integration?"

Because FastAPI auto-generates OpenAPI docs (/docs, /openapi.json) from your type hints, this is often a five-minute request rather than a day of writing documentation by hand — but only if you were disciplined about typing every request/response model. This is a concrete, visible payoff of the discipline this chapter pushes for.

---

Key takeaways

• FastAPI's validation, DI, and docs all derive from the same type hints — discipline there pays off everywhere.
• Sync vs async route handlers have materially different concurrency behavior.
• dependency_overrides turns the testability claim of Depends() into a real, working test.

Things you should be able to explain

• The ASGI request lifecycle through Uvicorn → Starlette → FastAPI → handler.
• Why Depends() with yield is the standard pattern for scoped resources like DB sessions.
• Why BackgroundTasks is not a durable job queue.

Things you should be able to build

• A FastAPI app with lifespan-managed shared clients, auth as a dependency, and a consistent response/error contract.
• A document-upload endpoint that validates type and size before parsing, and a test that swaps a real dependency for a fake via dependency_overrides.

Common mistakes

• Rebuilding clients per-request instead of at startup.
• Missing response_model, leaking internal fields.
• No rate limiting on LLM-triggering endpoints.
• Treating BackgroundTasks as a reliable queue for important work.
• Parsing an uploaded file before validating its type/size.

Recommended next chapter

04-sql.md