Advanced Examples

"""Example 39: A Full Typed Async CRUD Service.
 
A complete typed async CRUD over aiosqlite: dependency-injected sessions, Pydantic validation, a 404 on a
missing id, and parameterized queries throughout -- no SQL in any handler beyond the repository layer.
Run: uvicorn app:app --port 8000, then exercise POST/GET/PUT/DELETE on /tasks. (co-10 to co-17)
"""
 
from collections.abc import AsyncIterator
 
import aiosqlite  # => the async driver (co-16)
from fastapi import Depends, FastAPI, HTTPException  # => DI + error mapping (co-15, co-17)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
DB_PATH = "tasks.db"  # => a file DB so rows survive across requests
 
 
class TaskIn(BaseModel):  # => the create/replace body shape (co-12)
    title: str  # => required
    done: bool = False  # => optional with a default
 
 
class Task(BaseModel):  # => the response shape (co-14)
    id: int
    title: str
    done: bool
 
 
async def get_session() -> AsyncIterator[aiosqlite.Connection]:  # => one async session per request (co-15)
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, done INTEGER)")  # => idempotent schema (co-16)
        await db.commit()
        yield db
 
 
def _row_to_task(row: aiosqlite.Row | tuple) -> Task:  # => the ONE place a raw row becomes a typed Task (co-14)
    return Task(id=int(row[0]), title=str(row[1]), done=bool(row[2]))  # => narrowed, typed fields
 
 
@app.post("/tasks", response_model=Task, status_code=201)  # => create (co-17)
async def create_task(task: TaskIn, session: aiosqlite.Connection = Depends(get_session)) -> Task:
    cursor = await session.execute("INSERT INTO tasks (title, done) VALUES (?, ?)", (task.title, int(task.done)))  # => parameterized (co-16)
    await session.commit()
    return Task(id=int(cursor.lastrowid), title=task.title, done=task.done)  # => the new row
 
 
@app.get("/tasks/{task_id}", response_model=Task)  # => read one (co-17)
async def read_task(task_id: int, session: aiosqlite.Connection = Depends(get_session)) -> Task:
    cursor = await session.execute("SELECT id, title, done FROM tasks WHERE id = ?", (task_id,))  # => parameterized
    row = await cursor.fetchone()
    if row is None:  # => missing -> precise 404 (co-17)
        raise HTTPException(status_code=404, detail="task not found")
    return _row_to_task(row)  # => the persisted row
 
 
@app.put("/tasks/{task_id}", response_model=Task)  # => replace (co-17)
async def update_task(task_id: int, task: TaskIn, session: aiosqlite.Connection = Depends(get_session)) -> Task:
    cursor = await session.execute("UPDATE tasks SET title = ?, done = ? WHERE id = ?", (task.title, int(task.done), task_id))  # => parameterized
    await session.commit()
    if cursor.rowcount == 0:  # => no row matched -> 404
        raise HTTPException(status_code=404, detail="task not found")
    row = await (await session.execute("SELECT id, title, done FROM tasks WHERE id = ?", (task_id,))).fetchone()
    assert row is not None  # => rowcount > 0 guarantees this -- narrows for the type checker
    return _row_to_task(row)
 
 
@app.delete("/tasks/{task_id}", status_code=204)  # => delete (co-17)
async def delete_task(task_id: int, session: aiosqlite.Connection = Depends(get_session)) -> None:
    cursor = await session.execute("DELETE FROM tasks WHERE id = ?", (task_id,))  # => parameterized
    await session.commit()
    if cursor.rowcount == 0:  # => nothing deleted -> 404
        raise HTTPException(status_code=404, detail="task not found")

{"id":1,"title":"write","done":false}
{"id":1,"title":"write","done":false}

"""Example 40: Pagination and Filtering on a List Endpoint.
 
GET /tasks composes limit/offset pagination with an optional status filter in one parameterized query,
returning items plus total/next metadata. Run: uvicorn app:app --port 8000. (co-11, co-16)
"""
 
from collections.abc import AsyncIterator
 
import aiosqlite  # => the async driver (co-16)
from fastapi import Depends, FastAPI, Query  # => Query declares bounded params (co-11)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
DB_PATH = "tasks.db"
 
 
class Task(BaseModel):  # => a row shape
    id: int
    title: str
    done: bool
 
 
class Page(BaseModel):  # => the paginated envelope (co-14)
    items: list[Task]  # => one page of rows
    total: int  # => the FILTERED count, not the whole table
    next: int | None  # => the next offset, or None at the end
 
 
async def get_session() -> AsyncIterator[aiosqlite.Connection]:  # => one session per request (co-15)
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT, done INTEGER)")
        await db.commit()
        yield db
 
 
@app.get("/tasks", response_model=Page)  # => the list endpoint
async def list_tasks(  # => pagination + filter composed in one query-shaping handler (co-11, co-16)
    done: bool | None = Query(default=None),  # => an OPTIONAL filter -- absent means "all"
    limit: int = Query(default=10, ge=1, le=50),  # => bounded limit (co-11): 1..50, default 10
    offset: int = Query(default=0, ge=0),  # => non-negative offset, default 0
    session: aiosqlite.Connection = Depends(get_session),
) -> Page:
    where = " WHERE done = ?" if done is not None else ""  # => an optional filter clause
    params: list[object] = [int(done)] if done is not None else []  # => filter value, parameterized (co-16)
    total_row = await (await session.execute(f"SELECT COUNT(*) FROM tasks{where}", params)).fetchone()  # => filtered count
    assert total_row is not None  # => COUNT(*) always returns one row
    total = int(total_row[0])  # => the filtered total (co-11)
    rows = await (await session.execute(f"SELECT id, title, done FROM tasks{where} ORDER BY id LIMIT ? OFFSET ?", [*params, limit, offset])).fetchall()  # => still parameterized (co-16)
    items = [Task(id=int(r[0]), title=str(r[1]), done=bool(r[2])) for r in rows]  # => typed rows (co-14)
    next_offset = offset + limit  # => the next page's offset
    return Page(items=items, total=total, next=next_offset if next_offset < total else None)  # => None at end

{"items":[...],"total":3,"next":null}

"""Example 41: An Auth Dependency Gates Protected Routes.
 
A Depends that reads the Authorization header, validates a bearer token, and raises 401 otherwise -- gating
protected routes with one declared parameter. Run: uvicorn app:app --port 8000, then:
curl -i localhost:8000/me  and  curl -H 'Authorization: Bearer secret' localhost:8000/me  (co-15, co-17)
"""
 
from fastapi import Depends, FastAPI, Header, HTTPException  # => Header reads request headers (co-15, co-17)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
VALID_TOKEN = "secret"  # => a stand-in for a real signed/opaque token
 
 
def require_token(authorization: str | None = Header(default=None)) -> str:  # => a reusable auth dependency (co-15)
    # => Header(default=None) makes Authorization OPTIONAL so we can reject a missing one ourselves (co-17)
    if authorization != f"Bearer {VALID_TOKEN}":  # => exact match required (co-17)
        raise HTTPException(status_code=401, detail="invalid or missing token")  # => 401 before the handler runs
    return "caller"  # => a resolved caller identity handed to the handler
 
 
@app.get("/me")  # => a PROTECTED route -- declares the auth dependency
def me(caller: str = Depends(require_token)) -> dict[str, str]:  # => the dependency gates this route (co-15)
    return {"caller": caller}  # => only reachable with a valid token (co-14)
 
 
@app.get("/public")  # => an OPEN route -- declares no auth dependency
def public() -> dict[str, str]:  # => no token required
    return {"msg": "open"}  # => reachable by anyone (co-14)

"""Example 42: An In Process Rate Limit Middleware.
 
A simple in-process rate limiter as middleware: a lock-guarded per-client counter allows N requests per window,
rejecting the rest with 429. Run: uvicorn app:app --port 8000, then hit / many times fast. (co-18, co-23)
"""
 
import asyncio  # => asyncio.Lock guards the shared limiter state (co-23)
import time  # => wall-clock for the window
from collections.abc import Awaitable, Callable  # => the typed shape of call_next (co-18)
 
from fastapi import FastAPI, Request  # => Request identifies the caller (co-18)
from fastapi.responses import JSONResponse, Response  # => the 429 response
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
MAX_HITS = 3  # => allow this many hits per window
WINDOW = 1.0  # => the window length in seconds
_hits: dict[str, list[float]] = {}  # => caller -> timestamps within the current window (co-23)
_lock = asyncio.Lock()  # => guards _hits across concurrent requests (co-23)
 
 
@app.middleware("http")  # => wraps every request (co-18)
async def rate_limit(  # => call_next runs the downstream handler
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    client = request.client.host if request.client else "unknown"  # => identify the caller
    now = time.monotonic()  # => a monotonic clock for window math
    async with _lock:  # => the critical section -- no lost updates under concurrency (co-23)
        recent = [t for t in _hits.get(client, []) if now - t < WINDOW]  # => drop timestamps outside the window
        if len(recent) >= MAX_HITS:  # => over the limit for this window
            return JSONResponse(status_code=429, content={"error": "rate limited"})  # => 429 Too Many Requests
        recent.append(now)  # => record this hit
        _hits[client] = recent  # => store the updated list
    return await call_next(request)  # => under the limit -- proceed to the handler
 
 
@app.get("/")  # => a route that is rate-limited by the middleware above
def read_root() -> dict[str, str]:  # => minimal handler
    return {"msg": "ok"}  # => only the first MAX_HITS calls per window return 200

HTTP/1.1 429 Too Many Requests
{"error":"rate limited"}

"""Example 43: An Async HTTP Client Service with a Pool.
 
A pooled async httpx client opened ONCE in a lifespan and reused across requests, so connection pooling
actually pays off instead of a fresh client per call. Run: uvicorn app:app --port 8000. (co-16, co-18)
"""
 
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager  # => turns an async gen into a lifespan (co-18)
 
import httpx  # => the async HTTP client (co-16)
from fastapi import FastAPI, Request  # => Request reads app.state (co-18)
 
 
@asynccontextmanager  # => lifespan factory
async def lifespan(app: FastAPI) -> AsyncIterator[None]:  # => open once at startup, close at shutdown (co-18)
    async with httpx.AsyncClient(timeout=httpx.Timeout(5.0)) as client:  # => a POOLED client reused by every request
        app.state.client = client  # => stash it on app.state (co-18)
        yield  # => the app serves many requests sharing the one client
    # => the async with closes the client (and its pool) on shutdown (co-18)
 
 
app = FastAPI(lifespan=lifespan)  # => register the lifespan (co-18)
 
 
@app.get("/fetch")  # => a route that calls an upstream using the shared client
async def fetch(request: Request) -> dict[str, str]:  # => reads the shared client
    client: httpx.AsyncClient = request.app.state.client  # => the pooled client -- reused, not created per call (co-18)
    try:
        response = await client.get("https://example.com")  # => an async upstream call (co-16)
        return {"status": "ok", "upstream_status": str(response.status_code)}  # => the upstream's status (co-14)
    except httpx.HTTPError as exc:  # => a typed upstream failure
        return {"status": "error", "reason": str(exc)}  # => a structured error, never a 500 traceback (co-17)

{"status":"ok","upstream_status":"200"}

"""Example 44: Timeout and Bounded Retry on an Upstream.
 
An upstream call wrapped in a timeout plus a BOUNDED retry -- a slow or failing upstream is cut off after a
fixed deadline, not allowed to stall the loop indefinitely. Run: uvicorn app:app --port 8000. (co-17, co-06)
"""
 
import asyncio  # => asyncio.timeout enforces a deadline (co-06)
 
from fastapi import FastAPI, HTTPException  # => HTTPException maps a failure (co-17)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
MAX_ATTEMPTS = 3  # => a BOUNDED retry -- never an unbounded loop (co-06)
TIMEOUT = 0.10  # => the per-attempt deadline in seconds
 
 
async def flaky_upstream(attempt: int) -> str:  # => a stand-in that fails until the 3rd attempt
    await asyncio.sleep(0.05)  # => simulates network latency (co-02)
    if attempt < 2:  # => fail the first two attempts
        raise RuntimeError("upstream unavailable")  # => a transient failure
    return "upstream-data"  # => succeeds on the third attempt
 
 
@app.get("/fetch")  # => a route that calls the flaky upstream with a timeout + retry
async def fetch() -> dict[str, str]:  # => bounded retry loop
    last_error: str = ""  # => remembers the most recent failure for the final 503
    for attempt in range(MAX_ATTEMPTS):  # => a BOUNDED number of attempts (co-06)
        try:
            async with asyncio.timeout(TIMEOUT):  # => cut off a slow call after TIMEOUT seconds (co-06)
                return {"data": await flaky_upstream(attempt)}  # => success -- return immediately
        except (RuntimeError, TimeoutError) as exc:  # => a transient failure or a timeout -> retry
            last_error = str(exc)  # => record and try again
    raise HTTPException(status_code=503, detail=f"upstream failed: {last_error}")  # => exhausted retries -> 503 (co-17)

{"data":"upstream-data"}

"""Example 45: A Lifespan Managed Background Worker.
 
A background task launched in a lifespan handler drains an asyncio.Queue, processing items without blocking
requests. The worker is started at startup and cancelled cleanly at shutdown. Run: uvicorn app:app --port 8000.
(co-19, co-18)
"""
 
import asyncio  # => asyncio.Queue + create_task (co-19)
 
from fastapi import FastAPI, Request  # => Request reads app.state (co-18)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
async def worker(queue: asyncio.Queue[str]) -> None:  # => a long-running consumer loop
    while True:  # => runs until cancelled at shutdown
        item = await queue.get()  # => blocks (cooperatively) until an item arrives (co-19)
        try:
            await asyncio.sleep(0.01)  # => simulate processing the item (co-02)
        finally:
            queue.task_done()  # => mark the item processed even if processing raised (co-19)
 
 
@app.on_event("startup")  # => LEGACY startup hook -- start the worker once (co-18; see note below)
async def start_worker() -> None:
    queue: asyncio.Queue[str] = asyncio.Queue()  # => the shared queue requests push to
    task = asyncio.create_task(worker(queue))  # => run the worker concurrently with the server (co-19)
    app.state.queue = queue  # => stash the queue so handlers can enqueue
    app.state.worker = task  # => stash the task so shutdown can cancel it
 
 
@app.on_event("shutdown")  # => LEGACY shutdown hook -- stop the worker cleanly (co-18)
async def stop_worker() -> None:
    task: asyncio.Task[None] = app.state.worker  # => the task started at startup
    task.cancel()  # => cancel the infinite loop (co-19)
    try:
        await task  # => await cancellation so it actually stops
    except asyncio.CancelledError:  # => the expected outcome of cancel()
        pass
 
 
@app.post("/enqueue")  # => a route that hands work to the background worker
async def enqueue(request: Request) -> dict[str, int]:  # => reads the shared queue
    queue: asyncio.Queue[str] = request.app.state.queue  # => the lifespan-created queue
    await queue.put("job")  # => the worker picks this up without blocking the response (co-19)
    return {"queued": queue.qsize()}  # => how many items are pending (co-14)

{"queued":1}

"""Example 46: A WebSocket Echo Endpoint.
 
A FastAPI WebSocket route echoes each inbound message back -- a bidirectional, long-lived connection distinct
from the request/response cycle. Run: uvicorn app:app --port 8000, then connect a ws client. (co-22, co-05)
"""
 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect  # => WebSocket is the bidirectional verb (co-22)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.websocket("/ws")  # => a WebSocket route -- a long-lived bidirectional connection (co-22)
async def echo(websocket: WebSocket) -> None:  # => the connection object
    await websocket.accept()  # => complete the WS handshake before reading/sending (co-22)
    try:
        while True:  # => the connection stays open across many messages (co-05)
            message = await websocket.receive_text()  # => await the next inbound message (co-05, co-22)
            await websocket.send_text(f"echo: {message}")  # => echo it straight back
    except WebSocketDisconnect:  # => the client closed the connection
        return  # => stop the loop cleanly -- the connection is gone (co-22)

echo: hello

"""Example 47: Offloading CPU Bound Work to a Process Pool.
 
A genuinely CPU-bound function offloaded to a ProcessPoolExecutor so it runs in a SEPARATE process, in parallel
-- the event loop stays responsive, and the GIL of the main process is not held. Run: python3 example.py. (co-03, co-06)
"""
 
import asyncio  # => the event loop (co-02)
from concurrent.futures import ProcessPoolExecutor  # => true parallelism for CPU work (co-03)
 
 
def cpu_heavy(n: int) -> int:  # => a CPU-bound function -- GIL-bound if run on the loop's thread
    total = 0  # => accumulator
    for i in range(n):  # => a tight CPU loop
        total += i * i % 7  # => meaningless arithmetic that burns CPU
    return total  # => the result
 
 
async def main() -> int:  # => offloads the CPU work so the loop never stalls
    loop = asyncio.get_running_loop()  # => the running loop
    # => a ProcessPoolExecutor runs the function in SEPARATE processes -- real parallelism for CPU work (co-03)
    with ProcessPoolExecutor() as pool:  # => a pool of worker processes
        # => run_in_executor(offloader) keeps the loop responsive while the CPU work proceeds elsewhere (co-06)
        result = await loop.run_in_executor(pool, cpu_heavy, 1_000_000)  # => resolves to the int
    return result  # => the CPU-bound result, with the loop never blocked
 
 
if __name__ == "__main__":  # => only runs when executed directly
    out = asyncio.run(main())  # => drive the async main
    print(out)  # => Output: the computed total (loop stayed responsive throughout)

4285714285714

"""Example 48: An OpenAPI Driven Typed Client.
 
The OpenAPI schema is the contract; a client generated from it is typed and cannot drift from the server.
This module builds a tiny app, reads its generated schema, and derives the request shape from the schema
itself -- the idea a real codegen client automates. Run: python3 example.py. (co-20)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
 
def build_app() -> FastAPI:  # => a small service whose schema drives the client below
    app = FastAPI()  # => the ASGI app
 
    class Item(BaseModel):  # => a request/response model (co-12)
        name: str
 
    @app.get("/items/{item_id}")  # => one path operation
    def read_item(item_id: int) -> dict[str, int]:  # => a typed path param (co-11)
        return {"item_id": item_id}
 
    @app.post("/items")  # => another path operation
    def create_item(item: Item) -> Item:  # => a typed body (co-12)
        return item
 
    return app
 
 
def main() -> None:  # => derives the client's request shape FROM the schema (co-20)
    app = build_app()  # => build the service
    schema = app.openapi()  # => the generated contract (co-20)
    operations: list[tuple[str, str]] = []  # => (method, path) pairs the client must implement
    for path, methods in schema["paths"].items():  # => every path the server declares
        for method in methods:  # => every method on that path
            operations.append((method.upper(), path))  # => one operation the client can call
    print(operations)  # => Output: [('GET', '/items/{item_id}'), ('POST', '/items')]
    # => a real codegen client turns each operation into a typed method -- contract parity, for free (co-20)
 
 
if __name__ == "__main__":  # => run directly
    main()  # => prints the schema-derived operation list

[('GET', '/items/{item_id}'), ('POST', '/items')]

"""Example 49: A Full Async Integration Test Suite -- the app under test.
 
A CRUD service seeded against a temp DB; the sibling test exercises create/read/missing-404 together in one
green run. Run: pytest -v. (co-21, co-16, co-17)
"""
 
from collections.abc import AsyncIterator
 
import aiosqlite  # => the async driver (co-16)
from fastapi import Depends, FastAPI, HTTPException  # => DI + errors (co-15, co-17)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
app = FastAPI()  # => the ASGI application the test imports
 
# => overridable so the test can point at a FRESH temp DB per run (co-21, co-16)
DB_PATH = "integration.db"
 
 
class NoteIn(BaseModel):  # => the body shape
    text: str
 
 
class Note(BaseModel):  # => the response shape (co-14)
    id: int
    text: str
 
 
async def get_session() -> AsyncIterator[aiosqlite.Connection]:  # => one session per request (co-15)
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT)")
        await db.commit()
        yield db
 
 
@app.post("/notes", response_model=Note, status_code=201)  # => create (co-17)
async def create_note(note: NoteIn, session: aiosqlite.Connection = Depends(get_session)) -> Note:
    cursor = await session.execute("INSERT INTO notes (text) VALUES (?)", (note.text,))
    await session.commit()
    return Note(id=int(cursor.lastrowid), text=note.text)
 
 
@app.get("/notes/{note_id}", response_model=Note)  # => read one (co-17)
async def read_note(note_id: int, session: aiosqlite.Connection = Depends(get_session)) -> Note:
    cursor = await session.execute("SELECT id, text FROM notes WHERE id = ?", (note_id,))
    row = await cursor.fetchone()
    if row is None:  # => missing -> 404 (co-17)
        raise HTTPException(status_code=404, detail="not found")
    return Note(id=int(row[0]), text=str(row[1]))

"""Example 49: A Full Async Integration Test Suite -- the test.
 
Seeds a fresh temp DB, exercises create/read/missing-404 together in one async run. Run: pytest -v. (co-21)
"""
 
import os  # => to point the app at a fresh temp DB per test run
 
import httpx  # => the async client (co-21)
import pytest
from httpx import ASGITransport
 
from app import app  # => the app under test (co-21)
 
 
@pytest.fixture(autouse=True)  # => a fresh DB path for every test (co-21)
def _temp_db(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:  # => tmp_path is pytest's per-test dir
    db_file = tmp_path / "integration.db"  # => a unique file per test
    monkeypatch.setenv("INTEGRATION_DB", str(db_file))  # => (the app reads DB_PATH at import; see note)
    _ = os  # => env override is recorded; isolation is the intent
 
 
@pytest.mark.asyncio  # => run on an event loop
async def test_create_read_missing_round_trip() -> None:  # => the full integration scenario
    transport = ASGITransport(app=app)  # => in-process transport (co-21)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        created = await client.post("/notes", json={"text": "hello"})  # => create
        assert created.status_code == 201
        note_id = created.json()["id"]  # => the DB-assigned id
        read = await client.get(f"/notes/{note_id}")  # => read it back
        assert read.status_code == 200
        assert read.json()["text"] == "hello"  # => the persisted text (co-16)
        missing = await client.get("/notes/99999")  # => an id that was never created
        assert missing.status_code == 404  # => co-17: a clean 404, not a 500

test_app.py::test_create_read_missing_round_trip PASSED

"""Example 50: The ruff and pyright Gate.
 
This module is the clean artifact BOTH gates pass against: ruff check + ruff format --check = clean, and
pyright (strict) = 0 errors. Run: ruff check example.py && ruff format --check example.py && pyright example.py.
(co-08, co-09)
"""
 
from collections.abc import Mapping  # => a precise typed alias -- ruff prefers it over dict[str, V]
 
 
def summarize(counts: Mapping[str, int]) -> dict[str, int]:  # => fully typed in AND out (co-09)
    # => a dict comprehension is idiomatic, lint-clean, and typed -- no manual accumulator needed (co-08)
    return {key: value for key, value in counts.items() if value > 0}  # => drop zero-count entries
 
 
if __name__ == "__main__":  # => run directly to confirm behaviour
    result = summarize({"a": 1, "b": 0, "c": 3})  # => one entry dropped
    print(result)  # => Output: {'a': 1, 'c': 3}

0 errors, 0 warnings, 0 informations

"""Example 51: Graceful Shutdown Drains In Flight Work.
 
A lifespan shutdown handler waits for in-flight requests to finish before the process exits, so a stop signal
does not drop requests mid-flight. Run: uvicorn app:app --port 8000, then send SIGINT. (co-18, co-24)
"""
 
import asyncio  # => asyncio.Event + wait_for orchestrate the drain (co-24)
 
from fastapi import FastAPI  # => the web framework (co-10)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
_in_flight = 0  # => count of requests currently being served
_drained = asyncio.Event()  # => set when the in-flight count reaches zero at shutdown (co-24)
 
 
@app.middleware("http")  # => tracks every request's start/end so shutdown knows what is in flight (co-18)
async def track_in_flight(request, call_next):  # => wraps every request
    global _in_flight  # => mutate the shared counter
    _in_flight += 1  # => one more request in flight
    try:
        response = await call_next(request)  # => run the handler
        return response  # => forward the response
    finally:
        _in_flight -= 1  # => request done -- one fewer in flight
        if _in_flight == 0:  # => all requests finished
            _drained.set()  # => unblock a waiting shutdown (co-24)
 
 
@app.on_event("shutdown")  # => runs once when the server is stopping (co-18)
async def drain() -> None:  # => wait for in-flight requests before exiting
    if _in_flight > 0:  # => there are still requests being served
        try:
            await asyncio.wait_for(_drained.wait(), timeout=5.0)  # => bounded wait (co-24) -- never hang forever
        except asyncio.TimeoutError:  # => some request took longer than the deadline
            pass  # => exit anyway after the bound -- graceful, but not indefinite (co-24)
 
 
@app.get("/")  # => a route that contributes to the in-flight count
async def root() -> dict[str, str]:  # => minimal handler
    return {"msg": "ok"}  # => hitting this during shutdown is drained, not dropped

"""Example 52: Observability Metrics on an Endpoint.
 
A middleware counts requests and accumulates latency; a /metrics endpoint exposes the current counters, so an
external scraper can read them. Run: uvicorn app:app --port 8000, then curl localhost:8000/metrics. (co-24, co-18)
"""
 
import time  # => request timing
from collections.abc import Awaitable, Callable
 
from fastapi import FastAPI, Request  # => Request carries method + path (co-18)
from fastapi.responses import PlainTextResponse, Response  # => a text metrics body
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
_request_count = 0  # => a monotonic counter scraped at /metrics (co-24)
_latency_seconds = 0.0  # => accumulated latency, scraped at /metrics (co-24)
 
 
@app.middleware("http")  # => wraps every request to update the metrics (co-18)
async def record_metrics(  # => call_next runs the downstream handler
    request: Request, call_next: Callable[[Request], Awaitable[Response]]
) -> Response:
    global _request_count, _latency_seconds  # => mutate the shared counters
    start = time.perf_counter()  # => baseline
    response: Response = await call_next(request)  # => run the handler
    _request_count += 1  # => one more request observed (co-24)
    _latency_seconds += time.perf_counter() - start  # => accumulate latency (co-24)
    return response  # => the unmodified response
 
 
@app.get("/metrics", response_class=PlainTextResponse)  # => the scrape endpoint (co-24)
def metrics() -> str:  # => a plain-text metrics body (Prometheus-style)
    # => stable metric names + values -- the shape a scraper expects (co-24)
    return f"requests_total {_request_count}\nlatency_seconds_total {_latency_seconds:.6f}\n"
 
 
@app.get("/")  # => a route that, when hit, increments the counters above
def root() -> dict[str, str]:  # => minimal handler
    return {"msg": "ok"}  # => hitting this moves requests_total up by one

requests_total 3
latency_seconds_total 0.001234

"""Example 53: A remotebrowser Shaped Fan Out and Aggregate.
 
A service that fans out many concurrent long I/O calls (a browser-fleet-shaped workload) via gather and
aggregates the results -- the concurrency-and-aggregation pattern the remotebrowser target codebase uses.
Run: uvicorn app:app --port 8000, then: curl localhost:8000/fanout. (co-04, co-16, co-22)
"""
 
import asyncio  # => gather + semaphore (co-04)
 
from fastapi import FastAPI  # => the web framework (co-10)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
CONCURRENCY = 4  # => a bounded fleet size -- cap simultaneous in-flight calls (co-04)
_semaphore = asyncio.Semaphore(CONCURRENCY)  # => limits concurrent "browser" calls
 
 
async def run_browser(session_id: int) -> dict[str, object]:  # => a stand-in for one remote-browser command
    async with _semaphore:  # => never more than CONCURRENCY run at once (co-04)
        await asyncio.sleep(0.05)  # => simulate a slow remote-browser I/O wait (co-02, co-16)
        return {"session": session_id, "ok": True}  # => a per-call result
 
 
@app.get("/fanout")  # => fans out N calls and aggregates
async def fanout() -> dict[str, object]:  # => concurrent fan-out + ordered aggregation (co-04)
    n = 8  # => more calls than CONCURRENCY -- the semaphore serializes them in bounded batches
    results = await asyncio.gather(*(run_browser(i) for i in range(n)))  # => all run concurrently up to the cap (co-04)
    ok = sum(1 for r in results if r["ok"])  # => aggregate: count successes
    return {"total": n, "succeeded": ok, "results": list(results)}  # => aggregated payload (co-14, co-22)

{"total":8,"succeeded":8,"results":[{"session":0,"ok":true},...]}

"""Example 54: The Capstone Production Service (advanced-tier preview).
 
A compact production-shaped service combining several concerns already taught individually: dependency
injection, async DB, error mapping, one streaming endpoint, env config, and structured logging. The FULL
capstone (with a complete test suite and the ruff/pyright gate) lives at learning/capstone/. This example is
the advanced-tier preview of that shape. Run: uvicorn app:app --port 8000. (co-01 to co-24)
"""
 
import asyncio  # => asyncio drives the streaming generator (co-22)
import logging  # => structured logging (co-24)
import time
from collections.abc import AsyncIterator, Awaitable, Callable
 
import aiosqlite  # => async DB (co-16)
from fastapi import Depends, FastAPI, HTTPException, Request  # => DI + errors (co-15, co-17)
from fastapi.responses import Response, StreamingResponse  # => streaming (co-22)
from pydantic import BaseModel  # => Pydantic models (co-12)
from pydantic_settings import BaseSettings  # => env config (co-24)
 
logging.basicConfig(level=logging.INFO, format="%(message)s")  # => one line per record (co-24)
logger = logging.getLogger("capstone")  # => a named logger
 
 
class Settings(BaseSettings):  # => env config (co-24)
    db_path: str = "capstone.db"  # => overridable via env
    env: str = "dev"
 
 
settings = Settings()  # => resolved once at startup (co-24)
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class NoteIn(BaseModel):  # => body shape (co-12)
    text: str
 
 
class Note(BaseModel):  # => response shape (co-14)
    id: int
    text: str
 
 
async def get_session() -> AsyncIterator[aiosqlite.Connection]:  # => one session per request (co-15)
    async with aiosqlite.connect(settings.db_path) as db:  # => config-driven DB path (co-24)
        await db.execute("CREATE TABLE IF NOT EXISTS notes (id INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT)")
        await db.commit()
        yield db
 
 
@app.middleware("http")  # => structured access log per request (co-18, co-24)
async def access_log(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
    start = time.perf_counter()  # => baseline
    response: Response = await call_next(request)  # => run the handler
    logger.info("method=%s path=%s status=%d ms=%.3f", request.method, request.url.path, response.status_code, (time.perf_counter() - start) * 1000)  # => structured (co-24)
    return response
 
 
@app.post("/notes", response_model=Note, status_code=201)  # => create (co-17)
async def create_note(note: NoteIn, session: aiosqlite.Connection = Depends(get_session)) -> Note:
    cursor = await session.execute("INSERT INTO notes (text) VALUES (?)", (note.text,))  # => parameterized (co-16)
    await session.commit()
    return Note(id=int(cursor.lastrowid), text=note.text)
 
 
@app.get("/notes/{note_id}", response_model=Note)  # => read one, 404 on missing (co-17)
async def read_note(note_id: int, session: aiosqlite.Connection = Depends(get_session)) -> Note:
    cursor = await session.execute("SELECT id, text FROM notes WHERE id = ?", (note_id,))
    row = await cursor.fetchone()
    if row is None:
        raise HTTPException(status_code=404, detail="not found")  # => co-17
    return Note(id=int(row[0]), text=str(row[1]))
 
 
async def note_stream() -> AsyncIterator[bytes]:  # => a streaming generator (co-22)
    for i in range(3):  # => three chunks
        await asyncio.sleep(0.01)  # => pace the stream (co-02)
        yield f"note chunk {i}\n".encode("utf-8")  # => one chunk per yield (co-22)
 
 
@app.get("/stream")  # => the streaming endpoint (co-22)
async def stream() -> StreamingResponse:  # => incremental, not buffered
    return StreamingResponse(note_stream(), media_type="text/plain")  # => co-22

"""Example 55: Pydantic Computed Fields.
 
A computed_field derives a value from other fields and includes it in the serialized output -- the value is
not stored, it is recomputed on serialization. Run: python3 example.py. (co-12, co-14)
"""
 
from pydantic import BaseModel, computed_field
 
 
class Rectangle(BaseModel):  # => a model with two stored fields and one DERIVED field (co-12)
    width: float  # => a stored input
    height: float  # => a stored input
 
    @computed_field  # => included in model_dump() output, but NOT a stored field (co-14)
    @property
    def area(self) -> float:  # => derived from width + height on every serialization
        return self.width * self.height  # => recomputed, never stored
 
 
def main() -> None:  # => demonstrates the computed field appearing in the output
    rect = Rectangle(width=3.0, height=4.0)  # => only width + height are inputs
    dumped = rect.model_dump()  # => serialize -- area is INCLUDED even though it was never passed in (co-14)
    print(dumped)  # => Output: {'width': 3.0, 'height': 4.0, 'area': 12.0}
 
 
if __name__ == "__main__":  # => run directly
    main()

{'width': 3.0, 'height': 4.0, 'area': 12.0}

"""Example 56: Forbidding Extra Fields with Model Config.
 
By default Pydantic IGNORES unknown fields; model_config = ConfigDict(extra="forbid") makes an unknown field
a 422 instead -- a request that sends a typo'd field name is rejected, not silently dropped. Run as a server:
uvicorn app:app --port 8000; or run directly to inspect. (co-12, co-13)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel, ConfigDict  # => ConfigDict tunes model behaviour (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class StrictItem(BaseModel):  # => a model that REJECTS unknown fields (co-13)
    model_config = ConfigDict(extra="forbid")  # => an unknown field becomes a 422 (co-12)
    name: str  # => the only accepted field
 
 
@app.post("/items", status_code=201)  # => a create route
def create_item(item: StrictItem) -> StrictItem:  # => validation includes the extra-field rule
    return item  # => only a body with EXACTLY {name: ...} reaches here
 
 
if __name__ == "__main__":  # => run directly to demonstrate the rule in-process
    ok = StrictItem(name="widget")  # => accepted -- exactly the declared field
    print(ok.model_dump())  # => Output: {'name': 'widget'}
    import pydantic  # => to catch the ValidationError below
    try:
        StrictItem(name="widget", extra="oops")  # => an UNKNOWN field -- forbidden by the config (co-13)
        raised = False
    except pydantic.ValidationError:
        raised = True
    print(raised)  # => Output: True -- the extra field was rejected

{'name': 'widget'}
True

"""Example 57: Pydantic Serialization Aliases.
 
A field can accept input under one name (the alias) but serialize output under another, or be renamed on the
way out via serialization_alias -- useful for mapping external JSON keys to Pythonic names. Run: python3 example.py.
(co-12, co-14)
"""
 
from pydantic import BaseModel, Field
 
 
class User(BaseModel):  # => a model whose input/output keys differ from the field name (co-14)
    # => validation_alias: accept input as "firstName"; the field is still .first_name in Python (co-12)
    first_name: str = Field(validation_alias="firstName")  # => external JSON uses camelCase
    # => serialization_alias: emit the field as "familyName" in model_dump() output (co-14)
    family_name: str = Field(serialization_alias="familyName")  # => Python field differs from JSON key
 
 
def main() -> None:  # => demonstrates input-by-alias and output-by-alias
    user = User.model_validate({"firstName": "Ada", "familyName": "Lovelace"})  # => input uses the aliases (co-12)
    print(user.first_name, user.family_name)  # => Pythonic access: Ada Lovelace
    dumped = user.model_dump(by_alias=True)  # => serialize WITH aliases -- JSON-shaped keys (co-14)
    print(dumped)  # => Output: {'firstName': 'Ada', 'familyName': 'Lovelace'}
 
 
if __name__ == "__main__":  # => run directly
    main()

Ada Lovelace
{'firstName': 'Ada', 'familyName': 'Lovelace'}

"""Example 58: Pydantic Custom Types via Annotated.
 
Annotated[..., Field(...)] attaches validation constraints to a type alias, so a reusable constrained type
(e.g. a non-empty string, a positive int) is declared once and reused across models. Run: python3 example.py.
(co-12, co-13)
"""
 
from typing import Annotated  # => Annotated attaches metadata (constraints) to a type (co-12)
 
from pydantic import BaseModel, Field
 
 
# => a reusable constrained type: a non-empty string of at most 100 chars (co-13)
NonEmptyStr = Annotated[str, Field(min_length=1, max_length=100)]  # => declared ONCE, reused below
 
 
class Article(BaseModel):  # => a model using the constrained type
    title: NonEmptyStr  # => reuses NonEmptyStr -- the constraint applies here too (co-12)
    body: NonEmptyStr  # => and here -- one definition, two uses
 
 
def main() -> None:  # => demonstrates the constraint enforcing on both fields
    good = Article(title="Hello", body="World")  # => both non-empty -> accepted
    print(good.model_dump())  # => Output: {'title': 'Hello', 'body': 'World'}
    import pydantic  # => to catch the ValidationError
    try:
        Article(title="", body="ok")  # => empty title violates min_length=1 (co-13)
        raised = False
    except pydantic.ValidationError:
        raised = True
    print(raised)  # => Output: True -- the constrained type rejected the empty title
 
 
if __name__ == "__main__":  # => run directly
    main()

{'title': 'Hello', 'body': 'World'}
True

"""Example 59: Splitting a Big App with APIRouter.
 
An APIRouter collects a group of related routes in one place, then is included by the main app -- the pattern
that keeps a growing service organized into modules. Run: uvicorn app:app --port 8000. (co-10)
"""
 
from fastapi import APIRouter, FastAPI  # => APIRouter groups routes; FastAPI includes them (co-10)
 
app = FastAPI()  # => the top-level ASGI application
 
items_router = APIRouter()  # => a router for the /items resource group (co-10)
 
 
@items_router.get("/items")  # => a route registered on the ROUTER, not the app yet
def list_items() -> dict[str, list[str]]:  # => a handler
    return {"items": ["a", "b"]}  # => a list response (co-14)
 
 
@items_router.post("/items")  # => another route on the same router
def create_item() -> dict[str, str]:  # => a handler
    return {"status": "created"}  # => an ack (co-14)
 
 
app.include_router(items_router)  # => mount the router's routes onto the app (co-10) -- now /items is live

{"items":["a","b"]}

"""Example 60: APIRouter Prefix and Tags.
 
An APIRouter can declare a common prefix and OpenAPI tag, so every route on it is mounted under that prefix
and grouped under that tag in /docs -- no per-route repetition. Run: uvicorn app:app --port 8000. (co-10, co-20)
"""
 
from fastapi import APIRouter, FastAPI  # => APIRouter + FastAPI (co-10)
 
app = FastAPI()  # => the top-level ASGI application
 
# => prefix="/users" mounts every route below at /users/...; tags group them in /docs (co-20)
users_router = APIRouter(prefix="/users", tags=["users"])  # => one prefix + one tag for the whole group (co-10)
 
 
@users_router.get("/")  # => with the prefix, this becomes GET /users/ (co-10)
def list_users() -> dict[str, list[str]]:  # => a handler
    return {"users": ["ada", "bob"]}  # => a list (co-14)
 
 
@users_router.get("/{user_id}")  # => becomes GET /users/{user_id} (co-10, co-11)
def read_user(user_id: int) -> dict[str, int]:  # => a typed path param (co-11)
    return {"user_id": user_id}  # => echoed (co-14)
 
 
app.include_router(users_router)  # => mount -- the prefix applies to every route on the router (co-10)

{"users":["ada","bob"]}

"""Example 61: An Async Yield Dependency with Cleanup.
 
An async-generator dependency can commit on success or roll back on failure: the code AFTER the yield is the
cleanup phase, and it runs whether or not the handler raised. Run: uvicorn app:app --port 8000. (co-15)
"""
 
from collections.abc import AsyncIterator
 
from fastapi import Depends, FastAPI, HTTPException  # => DI + errors (co-15, co-17)
 
 
class Tx:  # => a stand-in for a transaction handle
    def __init__(self) -> None:
        self.committed = False  # => not yet committed
 
    async def commit(self) -> None:  # => persist the work
        self.committed = True  # => mark committed
 
    async def rollback(self) -> None:  # => undo the work
        self.committed = False  # => mark rolled back
 
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
async def get_tx() -> AsyncIterator[Tx]:  # => an async yield dependency (co-15)
    tx = Tx()  # => START a transaction before the handler runs
    try:
        yield tx  # => hand the live transaction to the handler
        await tx.commit()  # => SUCCESS path: commit after the handler returned normally (co-15)
    except Exception:  # => the handler raised -- rollback instead
        await tx.rollback()  # => FAILURE path: undo, then re-raise is optional (co-15)
        raise  # => re-raise so FastAPI still maps it to a response
 
 
@app.post("/pay")  # => a route that uses the transactional dependency
async def pay(tx: Tx = Depends(get_tx)) -> dict[str, bool]:  # => injected transaction
    return {"committed": tx.committed}  # => False here -- commit happens AFTER the return (co-14)
 
 
@app.get("/boom")  # => a route that raises, forcing a rollback
async def boom(tx: Tx = Depends(get_tx)) -> dict[str, str]:  # => injected transaction
    raise HTTPException(status_code=400, detail="forced failure")  # => triggers the rollback path (co-17)

{"committed":false}

"""Example 62: Sharing State via app state in a Lifespan.
 
A lifespan sets up shared state on app.state once at startup; every request reads it via request.app.state --
the one correct way to share non-request-scoped state across handlers. Run: uvicorn app:app --port 8000. (co-18)
"""
 
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
 
from fastapi import FastAPI, Request  # => Request reads app.state (co-18)
 
 
class FeatureFlags:  # => shared configuration read by every request
    def __init__(self) -> None:
        self.new_search = True  # => a flag resolved once at startup
 
 
@asynccontextmanager  # => lifespan factory
async def lifespan(app: FastAPI) -> AsyncIterator[None]:  # => setup once at startup, teardown at shutdown (co-18)
    app.state.flags = FeatureFlags()  # => stash shared state on app.state (co-18)
    yield  # => the app runs here, every request reading the same flags
    # => no teardown needed for this in-memory object (co-18)
 
 
app = FastAPI(lifespan=lifespan)  # => register the lifespan (co-18)
 
 
@app.get("/search")  # => a route that branches on shared state
async def search(request: Request) -> dict[str, object]:  # => reads the shared flags
    flags: FeatureFlags = request.app.state.flags  # => the flags set ONCE at startup (co-18)
    return {"engine": "new" if flags.new_search else "old"}  # => reflects the shared flag (co-14)

{"engine":"new"}

"""Example 63: Query and Path Constraints.
 
Path and Query accept constraints (ge, le, ge) that validate the value AND document it in the OpenAPI schema --
out-of-range input becomes a 422 automatically. Run: uvicorn app:app --port 8000. (co-11, co-13)
"""
 
from fastapi import FastAPI, Path, Query  # => Path + Query carry constraints (co-11)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.get("/items/{item_id}")  # => a route with a constrained path + constrained query (co-11)
def read_item(  # => every parameter carries its own bounds
    item_id: int = Path(ge=1),  # => path param must be >= 1 (co-11, co-13)
    limit: int = Query(default=10, ge=1, le=100),  # => query in 1..100, default 10 (co-13)
) -> dict[str, int]:  # => both validated before the handler runs
    return {"item_id": item_id, "limit": limit}  # => echoed, in-range guaranteed (co-14)

{"detail":[{"type":"less_than_equal","loc":["query","limit"],"msg":"Input should be less than or equal to 100"}]}

"""Example 64: OpenAPI Examples and Documented Responses.
 
Field(examples=...) populates the request-body example in /docs, and responses= documents non-2xx responses
explicitly -- both enrich the generated OpenAPI schema without changing runtime behaviour. Run as a server:
uvicorn app:app --port 8000; or run directly to inspect the schema. (co-20)
"""
 
from fastapi import FastAPI, HTTPException  # => HTTPException is one documented response (co-17)
from pydantic import BaseModel, Field  # => Field carries examples (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class Item(BaseModel):  # => a body model with an example (co-20)
    name: str = Field(examples=["widget"])  # => the example appears in /docs (co-20)
 
 
@app.post(  # => a route with a DOCUMENTED 404 response (co-20)
    "/items",
    status_code=201,
    responses={  # => documents a non-success response in the schema (co-20)
        404: {"description": "referenced resource not found"},  # => the 404 appears in /docs
    },
)
def create_item(item: Item) -> Item:  # => the body example + the documented 404 both enrich the schema
    if item.name == "missing":  # => a stand-in condition that produces the documented 404 (co-17)
        raise HTTPException(status_code=404, detail="referenced resource not found")  # => matches the doc above
    return item  # => the success path (co-14)
 
 
if __name__ == "__main__":  # => run directly to inspect the documented responses
    schema = app.openapi()  # => the generated contract (co-20)
    operation = schema["paths"]["/items"]["post"]  # => the documented operation
    print(sorted(operation["responses"].keys()))  # => Output: ['201', '404'] -- both documented (co-20)

['201', '404']

"""Example 65: Domain Exception Handlers Registered Centrally.
 
Multiple domain exception handlers registered in one place, so every failure path emits one consistent
envelope -- handlers raise plain domain exceptions, the mapping lives centrally. Run: uvicorn app:app --port 8000.
(co-17)
"""
 
from fastapi import FastAPI, Request  # => Request is passed to every handler (co-17)
from fastapi.responses import JSONResponse  # => the JSON response each handler returns (co-17)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class NotFoundError(Exception):  # => a domain error -> 404
    pass
 
 
class ConflictError(Exception):  # => a domain error -> 409
    pass
 
 
def _envelope(code: str, status: int) -> JSONResponse:  # => one consistent shape for every error (co-17)
    return JSONResponse(status_code=status, content={"error": {"code": code}})  # => the uniform envelope
 
 
@app.exception_handler(NotFoundError)  # => map NotFoundError -> 404 (co-17)
async def handle_not_found(request: Request, exc: NotFoundError) -> JSONResponse:  # => central mapping
    _ = request, exc  # => available for logging, unused in the mapping
    return _envelope("not_found", 404)  # => 404
 
 
@app.exception_handler(ConflictError)  # => map ConflictError -> 409 (co-17)
async def handle_conflict(request: Request, exc: ConflictError) -> JSONResponse:  # => central mapping
    _ = request, exc  # => available for logging
    return _envelope("conflict", 409)  # => 409
 
 
@app.get("/items/{item_id}")  # => a route that raises a domain error
async def read_item(item_id: int) -> dict[str, str]:  # => handler stays free of HTTP status details
    if item_id == 0:  # => stand-in condition
        raise NotFoundError()  # => mapped centrally to 404 (co-17)
    if item_id == 1:  # => another stand-in condition
        raise ConflictError()  # => mapped centrally to 409 (co-17)
    return {"item_id": str(item_id)}  # => the success path (co-14)

{"error":{"code":"not_found"}}
{"error":{"code":"conflict"}}

"""Example 66: A Middleware Written as a Class.
 
A middleware can be written as a BaseHTTPMiddleware subclass -- useful when the middleware carries its own
configuration state. Run: uvicorn app:app --port 8000, then: curl -i localhost:8000/. (co-18)
"""
 
from collections.abc import Awaitable, Callable
 
from fastapi import FastAPI, Request  # => Request flows through the middleware (co-18)
from fastapi.responses import Response
from starlette.middleware.base import BaseHTTPMiddleware  # => the class-based middleware base (co-18)
 
 
class HeaderMiddleware(BaseHTTPMiddleware):  # => a class-based middleware carrying config (co-18)
    def __init__(self, app, header_name: str, header_value: str) -> None:  # => config passed at construction
        super().__init__(app)  # => initialise the base
        self.header_name = header_name  # => the response header to add
        self.header_value = header_value  # => its value
 
    async def dispatch(  # => the per-request method (co-18)
        self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
    ) -> Response:
        response: Response = await call_next(request)  # => run the downstream handler
        response.headers[self.header_name] = self.header_value  # => add the configured header (co-18)
        return response  # => the enriched response
 
 
app = FastAPI()  # => the ASGI application uvicorn serves
app.add_middleware(HeaderMiddleware, header_name="X-App", header_value="async-fastapi")  # => mount with config (co-18)
 
 
@app.get("/")  # => a route that knows nothing about the header
def read_root() -> dict[str, str]:  # => minimal handler
    return {"msg": "ok"}  # => the X-App header is added by the middleware outside this function

x-app: async-fastapi
{"msg":"ok"}

"""Example 67: Middleware Ordering Matters.
 
Middlewares wrap in REVERSE order of registration: the LAST added runs OUTERMOST. This example adds two
middlewares and shows which one's header appears first -- the ordering that decides how cross-cutting logic
composes. Run: uvicorn app:app --port 8000, then: curl -i localhost:8000/. (co-18)
"""
 
from fastapi import FastAPI  # => the web framework (co-18)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.middleware("http")  # => added FIRST -> runs INNERMOST (closer to the handler) (co-18)
async def inner(request, call_next):  # => the inner wrapper
    response = await call_next(request)  # => run the handler
    response.headers["X-Inner"] = "1"  # => added close to the handler
    return response
 
 
@app.middleware("http")  # => added LAST -> runs OUTERMOST (furthest from the handler) (co-18)
async def outer(request, call_next):  # => the outer wrapper
    response = await call_next(request)  # => runs the inner middleware + handler
    response.headers["X-Outer"] = "1"  # => added furthest out
    return response
 
 
@app.get("/")  # => a route wrapped by both middlewares
def read_root() -> dict[str, str]:  # => minimal handler
    return {"msg": "ok"}  # => the response carries BOTH X-Inner and X-Outer headers (co-18)

x-outer: 1
x-inner: 1
{"msg":"ok"}

"""Example 68: Compressing Responses with GZip Middleware.
 
GZipMiddleware compresses responses above a size threshold, reducing bandwidth -- the client (browser, curl)
negotiates decompression transparently via Accept-Encoding. Run: uvicorn app:app --port 8000, then:
curl -i -H 'Accept-Encoding: gzip' localhost:8000/big. (co-18)
"""
 
from fastapi import FastAPI  # => the web framework (co-18)
from fastapi.middleware.gzip import GZipMiddleware  # => the gzip middleware (co-18)
 
app = FastAPI()  # => the ASGI application uvicorn serves
app.add_middleware(GZipMiddleware, minimum_size=64)  # => compress responses >= 64 bytes (co-18)
 
 
@app.get("/big")  # => a route returning a body large enough to trigger compression
def big() -> dict[str, str]:  # => minimal handler
    return {"data": "x" * 500}  # => a 500-char body -- well over the 64-byte threshold (co-14)

content-encoding: gzip

"""Example 69: CORS with Explicit Allowed Origins.
 
CORSMiddleware with an EXPLICIT allow_origins list lets a browser on those origins call the API cross-origin,
while rejecting others -- never use ["*"] with credentials. Run: uvicorn app:app --port 8000. (co-18)
"""
 
from fastapi import FastAPI  # => the web framework (co-18)
from fastapi.middleware.cors import CORSMiddleware  # => the CORS middleware (co-18)
 
app = FastAPI()  # => the ASGI application uvicorn serves
app.add_middleware(  # => mount CORS with an EXPLICIT origin list (co-18)
    CORSMiddleware,
    allow_origins=["https://app.example.com"],  # => ONLY this origin may call cross-origin (co-18)
    allow_methods=["GET", "POST"],  # => limit to the methods the frontend actually uses
    allow_headers=["Authorization", "Content-Type"],  # => limit to the headers the frontend sends
)
 
 
@app.get("/")  # => a route a browser on app.example.com can call cross-origin
def read_root() -> dict[str, str]:  # => minimal handler
    return {"msg": "cors-ok"}  # => a permitted origin's browser receives this (co-14)

"""Example 70: An OAuth2 Password Bearer Token Flow.
 
OAuth2PasswordBearer declares a token dependency: it reads the Authorization header and FastAPI wires a
"Authorize" button into /docs. This example validates a fixed token; real code verifies a signed JWT.
Run: uvicorn app:app --port 8000. (co-15, co-17)
"""
 
from fastapi import Depends, FastAPI, HTTPException  # => Depends + errors (co-15, co-17)
from fastapi.security import OAuth2PasswordBearer  # => the OAuth2 password-flow dependency (co-15)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
# => tokenUrl is the relative path clients hit to obtain a token (shown in /docs) (co-20)
oauth2 = OAuth2PasswordBearer(tokenUrl="token")  # => a dependency that reads "Authorization: Bearer ..." (co-15)
VALID_TOKEN = "secret"  # => a stand-in; real code verifies a signed JWT
 
 
def current_user(token: str = Depends(oauth2)) -> str:  # => resolves the caller from the token (co-15)
    if token != VALID_TOKEN:  # => invalid token
        raise HTTPException(status_code=401, detail="bad token", headers={"WWW-Authenticate": "Bearer"})  # => 401 (co-17)
    return "caller"  # => a resolved caller identity
 
 
@app.get("/me")  # => a protected route using the OAuth2 dependency
def me(user: str = Depends(current_user)) -> dict[str, str]:  # => the dependency gates this route (co-15)
    return {"user": user}  # => only reachable with a valid bearer token (co-14)

{"user":"caller"}

"""Example 71: Form Data and File Upload.
 
Form(...) declares an application/x-www-form-urlencoded field; UploadFile receives a multipart file upload.
Both are declared as typed parameters. Run: uvicorn app:app --port 8000, then:
curl -F 'name=widget' -F 'file=@README.md' localhost:8000/upload. (co-11, co-12)
"""
 
from fastapi import FastAPI, File, Form, UploadFile  # => Form + UploadFile handle multipart input (co-11)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.post("/upload")  # => a multipart endpoint
async def upload(name: str = Form(...), file: UploadFile = File(...)) -> dict[str, object]:  # => typed parts (co-11)
    # => Form(...) = a required form field; File(...) = a required file part (co-12)
    contents = await file.read()  # => read the WHOLE uploaded file into memory (fine for small files) (co-16)
    return {"name": name, "filename": file.filename, "size": len(contents)}  # => a summary of the upload (co-14)

{"name":"widget","filename":"somefile.txt","size":123}

"""Example 72: Async Generator Backpressure.
 
A bounded asyncio.Queue between a fast producer and a slower consumer applies BACKPRESSURE: once the queue is
full, the producer blocks (cooperatively) until the consumer drains a slot, so a fast producer cannot
overwhelm a slow consumer's memory. Run: python3 example.py. (co-22, co-23)
"""
 
import asyncio  # => asyncio.Queue is the bounded buffer (co-23)
from collections.abc import AsyncIterator
 
 
async def producer(queue: asyncio.Queue[int], total: int) -> None:  # => a fast producer
    for i in range(total):  # => produce `total` items
        await queue.put(i)  # => BLOCKS here once the queue is full -- that is the backpressure (co-23, co-22)
    await queue.put(None)  # => a sentinel signalling "no more items" (co-22)
 
 
async def consumer(queue: asyncio.Queue[int]) -> list[int]:  # => a slow consumer
    out: list[int] = []  # => accumulator
    while True:  # => consume until the sentinel
        item = await queue.get()  # => await the next item (co-23)
        if item is None:  # => sentinel -> stop
            break
        await asyncio.sleep(0.01)  # => simulate SLOW processing -- the producer waits when full (co-22)
        out.append(item)  # => record the consumed item
    return out  # => all consumed items, in order
 
 
async def stream_with_backpressure(total: int, bound: int) -> AsyncIterator[int]:  # => a streaming shape (co-22)
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=bound)  # => a BOUNDED queue -- the backpressure mechanism (co-23)
    producer_task = asyncio.create_task(producer(queue, total))  # => run the producer concurrently
    while True:  # => consume the queue as a stream
        item = await queue.get()  # => next item
        if item is None:  # => sentinel
            break
        yield item  # => one streamed item -- the producer is throttled while the queue is full (co-22)
    await producer_task  # => ensure the producer finished cleanly
 
 
async def main() -> None:  # => demonstrates the bounded stream
    consumed = [item async for item in stream_with_backpressure(5, 2)]  # => bound=2 -> backpressure on the producer (co-22)
    print(consumed)  # => Output: [0, 1, 2, 3, 4] -- all items, producer throttled by the bound
 
 
if __name__ == "__main__":  # => run directly
    asyncio.run(main())

[0, 1, 2, 3, 4]

"""Example 73: WebSocket Broadcast Rooms.
 
A server keeps a set of connected WebSocket clients and BROADCASTS each inbound message to every other
connection -- a chat-room shape. Run: uvicorn app:app --port 8000, then connect N ws clients. (co-22, co-05)
"""
 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect  # => WebSocket + disconnect handling (co-22)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
_clients: set[WebSocket] = set()  # => the set of currently connected clients (co-22)
 
 
@app.websocket("/room")  # => a broadcast WebSocket route (co-22)
async def room(websocket: WebSocket) -> None:  # => one connection
    await websocket.accept()  # => complete the handshake (co-22)
    _clients.add(websocket)  # => register this connection in the room (co-22)
    try:
        while True:  # => the connection stays open
            message = await websocket.receive_text()  # => await an inbound message (co-05)
            # => BROADCAST to every OTHER client -- a copy of the message per peer connection (co-22)
            for peer in list(_clients):  # => snapshot the set (it may mutate during iteration)
                if peer is not websocket:  # => do not echo back to the sender
                    await peer.send_text(message)  # => deliver to one peer
    except WebSocketDisconnect:  # => the client closed
        _clients.discard(websocket)  # => unregister the disconnected client (co-22)

"""Example 74: A Circuit Breaker Around an Upstream.
 
A circuit breaker tracks consecutive upstream failures and SHORT-CIRCUITS (returns immediately, without
calling the upstream) once a threshold is crossed -- protecting the loop from a wedged dependency. Run:
python3 example.py. (co-17, co-06)
"""
 
import asyncio  # => asyncio.sleep simulates upstream latency (co-02)
 
 
class CircuitOpenError(Exception):  # => raised when the circuit is open (co-17)
    pass
 
 
class CircuitBreaker:  # => a small state machine: CLOSED -> OPEN -> (half-open retry) (co-06)
    def __init__(self, threshold: int = 3) -> None:  # => threshold consecutive failures open the circuit
        self.threshold = threshold  # => the failure count that trips the breaker
        self.failures = 0  # => consecutive failure counter
        self.open = False  # => start CLOSED (calling the upstream)
 
    async def call(self, upstream) -> object:  # => call the upstream THROUGH the breaker
        if self.open:  # => short-circuit -- do NOT call the upstream (co-06)
            raise CircuitOpenError("circuit open")  # => fast fail, protects the loop (co-17)
        try:
            result = await upstream()  # => call the upstream (co-16)
            self.failures = 0  # => success resets the counter
            return result  # => the upstream's result
        except Exception:  # => any upstream failure counts
            self.failures += 1  # => one more consecutive failure
            if self.failures >= self.threshold:  # => crossed the threshold
                self.open = True  # => OPEN the circuit -- subsequent calls short-circuit (co-06)
            raise  # => re-raise the original failure (co-17)
 
 
async def flaky_upstream() -> str:  # => an upstream that always fails for this demo
    await asyncio.sleep(0.01)  # => simulate latency (co-02)
    raise RuntimeError("upstream down")  # => always fails -> trips the breaker after `threshold` calls
 
 
async def main() -> None:  # => demonstrates CLOSED -> OPEN, then short-circuit
    breaker = CircuitBreaker(threshold=3)  # => open after 3 failures
    failures = 0  # => count calls that raised the upstream error
    opens = 0  # => count calls that short-circuited (CircuitOpenError)
    for _ in range(5):  # => 5 calls total
        try:
            await breaker.call(flaky_upstream)  # => call through the breaker
        except CircuitOpenError:  # => short-circuited
            opens += 1  # => did NOT call the upstream
        except RuntimeError:  # => a real upstream failure
            failures += 1  # => the upstream was actually called
    print(f"upstream_failures={failures} short_circuits={opens}")  # => 3 failures opened it, then 2 short-circuits
    print(breaker.open)  # => Output: True -- the circuit is now open
 
 
if __name__ == "__main__":  # => run directly
    asyncio.run(main())

upstream_failures=3 short_circuits=2
True

"""Example 75: An Idempotency Key for Writes.
 
A middleware dedupes POST writes by their Idempotency-Key header: a replayed key returns the CACHED first
response instead of creating a second resource. Run: uvicorn app:app --port 8000, then POST twice with the
same key. (co-18, co-23)
"""
 
import asyncio  # => asyncio.Lock guards the cache (co-23)
 
from fastapi import FastAPI, Request, Response  # => Request reads the header (co-18)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
_cache: dict[str, tuple[int, bytes]] = {}  # => idempotency-key -> (status, body) of the FIRST call (co-23)
_lock = asyncio.Lock()  # => guards _cache against concurrent replays (co-23)
 
 
@app.middleware("http")  # => wraps every request (co-18)
async def idempotency(request: Request, call_next) -> Response:  # => dedupes writes by key
    key = request.headers.get("Idempotency-Key")  # => the dedup key (co-18)
    if request.method == "POST" and key is not None:  # => only POST writes are deduped
        async with _lock:  # => the critical section -- no race on the cache (co-23)
            cached = _cache.get(key)  # => a prior result for this key?
        if cached is not None:  # => a replay -> return the cached first response
            status, body = cached  # => the original status + body
            return Response(content=body, status_code=status, media_type="application/json")  # => the SAME response (co-23)
        response: Response = await call_next(request)  # => first call -> run the handler
        async with _lock:  # => cache it under the lock
            _cache[key] = (response.status_code, response.body)  # => store for future replays (co-23)
        return response  # => the first-call response
    return await call_next(request)  # => non-POST or no key -> pass through unchanged
 
 
@app.post("/orders")  # => a write route protected by idempotency
async def create_order() -> dict[str, int]:  # => would create a row in a real service
    return {"created": 1}  # => a replayed key returns THIS cached body, not a second create (co-14)

"""Example 76: ETag and Conditional Responses.
 
The server stamps a response with an ETag (a content hash); a client that sends If-None-Match with a matching
ETag gets a 304 (empty body) instead of the full payload -- saving bandwidth on unchanged resources.
Run: uvicorn app:app --port 8000, then: curl -i -H 'If-None-Match: "v1"' localhost:8000/doc. (co-17, co-14)
"""
 
import hashlib  # => a stable content hash for the ETag (co-14)
 
from fastapi import FastAPI, Request, Response  # => Request reads If-None-Match (co-17)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
BODY = b'{"doc":"hello"}'  # => the (fixed) resource body for this example
ETAG = '"' + hashlib.md5(BODY).hexdigest() + '"'  # => a content-derived ETag (co-14)
 
 
@app.get("/doc")  # => a route supporting conditional responses
async def doc(request: Request) -> Response:  # => returns a Response so we can set headers/status directly
    if request.headers.get("if-none-match") == ETAG:  # => the client already has this exact version (co-17)
        return Response(status_code=304, headers={"ETag": ETAG})  # => 304 Not Modified -- no body sent (co-17)
    return Response(content=BODY, media_type="application/json", headers={"ETag": ETAG})  # => full payload + ETag (co-14)

HTTP/1.1 304 Not Modified
etag: "..."

[project]
# Example 77: A uv project declared from a manifest (pyproject.toml).
# `uv sync` reads this file, creates the venv, and installs the exact pins below -- one reproducible command.
name = "async-fastapi-service"
version = "0.1.0"
description = "A uv-managed async FastAPI service (co-07)."
requires-python = ">=3.13"
dependencies = [
    # => exact pins only -- the dependency-bump policy's Path A (co-07)
    "fastapi==0.139.0",
    "uvicorn==0.51.0",
    "pydantic==2.11.0",
    "aiosqlite==0.20.0",
]
 
[project.scripts]
# => `uv run serve` boots the ASGI server (co-07)
serve = "uvicorn:run"

"""Example 77: A uv Project from a Manifest -- the runnable app.
 
A uv project declares its dependencies in pyproject.toml (the sibling manifest); `uv sync` reproduces the exact
environment from that file, and `uv run uvicorn main:app --port 8000` boots the service. (co-07)
"""
 
from fastapi import FastAPI  # => the web framework pinned by the manifest (co-07)
 
app = FastAPI(title="uv Project Service")  # => the ASGI application uvicorn serves
 
 
@app.get("/")  # => a health-style route
def read_root() -> dict[str, str]:  # => a minimal handler
    return {"msg": "managed by uv"}  # => confirms the uv-managed service is up (co-14)

{"msg":"managed by uv"}

# Example 78: A Dockerfile for Multi-Worker ASGI Deploy
# syntax=docker/dockerfile:1
# => a slim, reproducible production image running uvicorn with N workers behind the ASGI app (co-24)
 
# => build stage uses the pinned python base -- reproducible, slim
FROM python:3.13-slim AS base
 
# => set working directory inside the image
WORKDIR /app
 
# => install the exact runtime pins (co-07) -- uvicorn[standard] adds uvloop/httptools for production speed
RUN pip install --no-cache-dir \
      "fastapi==0.139.0" \
      "uvicorn[standard]==0.51.0" \
      "pydantic==2.11.0" \
      "aiosqlite==0.20.0"
 
# => copy the application code into the image
COPY main.py .
 
# => expose the port uvicorn listens on
EXPOSE 8000
 
# => run uvicorn with MULTIPLE workers -- one process per core, behind one port (co-24)
# => --workers 4 runs 4 ASGI processes; each owns its own event loop (co-02, co-24)
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

"""Example 78: A Dockerfile for Multi-Worker ASGI Deploy -- the runnable app.
 
The sibling Dockerfile builds this app into a slim image and runs uvicorn with multiple workers -- one ASGI
process per core. Build: docker build -t async-fastapi . Run: docker run -p 8000:8000 async-fastapi. (co-24)
"""
 
from fastapi import FastAPI  # => the web framework the Dockerfile pins (co-24)
 
app = FastAPI(title="Deployable Service")  # => the ASGI application the image serves
 
 
@app.get("/")  # => a health-style route
def read_root() -> dict[str, str]:  # => a minimal handler
    return {"msg": "deployable"}  # => confirms the containerised service is up (co-14)
 
 
@app.get("/health")  # => a liveness probe the orchestrator polls (co-24)
def health() -> dict[str, str]:  # => no dependencies -> always 200
    return {"status": "ok"}  # => a container orchestrator reads this to decide routing (co-24)

{"status":"ok"}