Overview

async def defines a coroutine function; calling it does not run the body -- it returns a coroutine object that you must schedule (usually by await-ing it or handing it to asyncio.run). The body only runs when something drives the coroutine on an event loop.

Cooperative. The loop runs one ready coroutine until it awaits, then switches to another -- it never preempts a running coroutine. So a coroutine that never yields (a blocking call) stalls every other coroutine on the same loop, not just itself.

No. async gives I/O concurrency on one core -- many in-flight waits share one thread. CPU-bound work still needs processes or threads (or the free-threaded build) to run in parallel; offloading it to a ProcessPoolExecutor is the pattern (ex-47). async alone buys nothing for CPU work.

create_task wraps a coroutine in a Task and schedules it to run concurrently immediately. gather runs several awaitables concurrently and returns their results in submission order (a list). Both turn "await one at a time" into "await many at once."

async with runs an async resource's __aenter__/__aexit__ (whose acquire/release themselves await) -- needed for connections and response bodies. async for pulls items from an async generator one at a time, awaiting between items -- needed for async streams. Both are for resources/streams whose operations are themselves awaitable.

It blocks the entire loop, not just that coroutine, because the loop cannot switch while the blocking call runs. Fix one: use an async-native equivalent (asyncio.sleep, an async driver). Fix two: offload genuinely blocking code to an executor via run_in_executor.

uv replaces the slower python -m venv + pip pair for creating environments and installing dependencies. A pinned uv pip install (or uv sync from a pyproject.toml) guarantees a reproducible environment -- the same exact versions across machines, CI, and deploys.

ruff check lints -- flags likely bugs and smells (unused imports, undefined names) without rewriting. ruff format formats -- rewrites code into one deterministic style (spacing, line breaks) without judging correctness. They are two complementary passes.

pyright type-checks Python against its type hints, catching contract errors (wrong argument type, None where an int is expected) before runtime. In a FastAPI service the type hints are the API contract, so pyright checks the contract in the editor before a single request runs.

The same typed signature drives validation, serialization, and the OpenAPI docs -- one annotation is simultaneously the validation rule, the response shape, and the schema entry. The route is where the framework reads the types and turns them into behaviour.

A parameter whose name appears in the route's {} placeholder is a path param; a Pydantic-model parameter is the body; any other (typically a scalar) is a query param. The type annotation on each is the validation rule.

It validates the incoming data against the declared field types and constraints, raising a ValidationError on a violation (a 422 in a handler). In a handler it is the boundary a request must cross before any handler logic runs, and the shape a response is filtered through on the way out.

Validation runs before the handler's own code executes. A failure (missing field, wrong type, violated constraint) produces a structured 422 naming the offending location -- the handler body only ever sees already-valid input.

It filters the return value to exactly the model's fields -- extra fields are stripped. So a field present on the internal object (a password hash) never reaches the response body if the output model omits it, regardless of what the handler returns.

Depends supplies a per-request resource (a DB session, a config object, an authenticated user) directly as a typed handler argument. The framework resolves the provider before the handler runs (and tears it down after), so the handler never knows how the resource was created.

A synchronous driver blocks the thread on each query, stalling the entire event loop (the co-06 hazard) and throwing away the concurrency the rest of the service depends on. An async driver (aiosqlite, asyncpg) lets each query wait yield to the loop, so concurrent requests overlap their query latency.

HTTPException(status_code=...) maps a single failure to a status + body in one call. A registered exception handler maps a whole class of domain errors to a response, so handlers raise plain domain exceptions and the central mapping decides the status and the consistent envelope shape.

Middleware wraps every request/response (cross-cutting behaviour: timing, logging, CORS). A lifespan handler runs once at startup and once at shutdown (pooled resources: a connection pool, an HTTP client, a background worker).

BackgroundTasks runs after the response is sent (a one-shot side effect). Reach for a lifespan-managed worker when the work needs a long-running consumer loop (a job processor, a metrics flusher) -- a BackgroundTask is one-shot, not continuous.

It is generated from the typed routes -- every path, parameter, model, and status derives from the same annotations that enforce the contract. So the docs cannot drift from the code: the schema and the implementation share one source of truth, which is what makes schema-driven client generation safe.

Via ASGITransport(app=app): the AsyncClient calls the ASGI app in-process -- a real request/response cycle with no socket and no running uvicorn. It is fast, deterministic, and part of the same dev loop as curl.

Streaming sends the first byte immediately and keeps peak memory flat (a buffered multi-megabyte response is slow to first byte and memory-hungry). StreamingResponse consumes an async generator lazily, flushing each yielded chunk to the client as it is produced.

False. The read-modify-write window between two awaits is exactly where a concurrent coroutine can interleave, so shared mutable state can still lose updates. The fix is an asyncio.Lock around the critical section, or better, avoiding shared state entirely.

Typed config from environment (pydantic-settings) so a missing value fails fast at startup; structured logging (stable keys, machine-parseable) instead of print; and an ASGI server config (host/port, workers, graceful-timeout) so the process runs behind multiple cores and shuts down cleanly.

requests.get is blocking -- it stalls the entire event loop for the duration of the call, queuing every concurrent request behind it (co-06). The fix is an async client (httpx.AsyncClient), awaited; if the blocking library is unavoidable, offload it with run_in_executor (co-03, ex-08).

count += 1 is a read-modify-write; the awaits around it let another coroutine interleave between the read and the write, losing updates (co-23). Wrap the critical section in an asyncio.Lock, or move the counter into the database (avoiding shared in-process state entirely).

Fan out the three calls with asyncio.gather so they run concurrently on one loop -- the handler pays the slowest call's latency (~100ms), not the sum (co-04, ex-28). Each upstream call must itself be async.

An async-generator Depends dependency that async withs the connection, yields it, and closes it in the finally (which runs even on error) -- co-15, co-05. Each handler just declares the dependency and receives an open session.

model_config = ConfigDict(extra="forbid") on the request model -- the unknown titel field becomes a 422 instead of being silently ignored (co-12, co-13, ex-56). The typo is then rejected loudly at the boundary.

Declare a response_model=UserOut on the route whose fields omit password_hash -- response_model filters the output to exactly that shape, stripping the hash regardless of what the handler returns (co-14, ex-17).

Open it in a lifespan handler, stash it on app.state, and close it after the yield (co-18). Handlers read it via request.app.state.pool (ex-25, ex-43, ex-62). Never open a pool per request -- it throws away pooling.

asyncio.timeout(deadline) cuts off a slow call after the deadline (co-06), and a bounded for loop retries on transient failure a fixed number of times. Exhaustion maps to a precise 503 (co-17, ex-44) -- never an unbounded retry loop.

Drive the app in-process with httpx.AsyncClient(ASGITransport(app=app)) (co-21) and give each test a FRESH temp DB file via a tmp_path fixture that overrides the config's DB path (ex-49, capstone). No socket, no shared state, deterministic order.

One consistent JSON error envelope (e.g. {"error": {"code", "message"}}) with the right status, never a raw traceback or internal detail. Register exception handlers so every failure path -- including unhandled ones -- emits that shape (co-17, ex-24, ex-65).

CORSMiddleware with an explicit allow_origins=["https://app.example.com"] -- that origin may call cross-origin, others are rejected. Never ["*"] with credentials (co-18, ex-69).

A type-and-lint gate: ruff check (lint + format check) AND pyright (strict) over the application code, run in CI as a blocking step (co-08, co-09, ex-50). Both, because ruff catches the bugs/style pyright cannot and pyright catches the contract errors ruff cannot.

import asyncio  # => co-04: gather runs the waits concurrently
import time
 
 
async def pause(seconds: float) -> None:  # => a unit of async work
    await asyncio.sleep(seconds)  # => yields to the loop (co-01)
 
 
async def measure_sequential(d: float) -> float:  # => two awaits IN SEQUENCE -> ~2d
    start = time.perf_counter()  # => baseline
    await pause(d)  # => first wait
    await pause(d)  # => second wait, stacked after the first
    return time.perf_counter() - start  # => ~2d
 
 
async def measure_concurrent(d: float) -> float:  # => gather -> ~d
    start = time.perf_counter()  # => baseline
    await asyncio.gather(pause(d), pause(d))  # => both run concurrently on one loop (co-04)
    return time.perf_counter() - start  # => ~d -- the max, not the sum
 
 
def main() -> None:  # => drives both measurements
    seq = asyncio.run(measure_sequential(0.10))  # => sequential
    con = asyncio.run(measure_concurrent(0.10))  # => concurrent
    print(f"sequential={seq:.3f}s concurrent={con:.3f}s")  # => sequential ~0.2, concurrent ~0.1
    assert con < seq  # => concurrency is genuinely faster here
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

sequential=0.201s concurrent=0.101s

import asyncio  # => co-06: blocking stalls the loop
import time
 
 
async def good(d: float) -> None:  # => an async wait
    await asyncio.sleep(d)  # => yields
 
 
async def blocker(d: float) -> None:  # => a BLOCKING wait -- the hazard (co-06)
    time.sleep(d)  # => NO await -> freezes the whole loop
 
 
async def fixed(d: float) -> None:  # => offloaded blocking -> loop stays responsive (co-06, co-03)
    loop = asyncio.get_running_loop()  # => the running loop
    await loop.run_in_executor(None, time.sleep, d)  # => blocking runs on a thread
 
 
async def run_with_blocker(d: float) -> float:  # => gather bad + good -> ~2d (bad froze good)
    start = time.perf_counter()
    await asyncio.gather(blocker(d), good(d))  # => bad blocks good
    return time.perf_counter() - start
 
 
async def run_fixed(d: float) -> float:  # => gather fixed + good -> ~d (concurrency restored)
    start = time.perf_counter()
    await asyncio.gather(fixed(d), good(d))  # => both run concurrently again
    return time.perf_counter() - start
 
 
def main() -> None:
    stalled = asyncio.run(run_with_blocker(0.10))  # => ~0.2 -- the stall
    ok = asyncio.run(run_fixed(0.10))  # => ~0.1 -- fixed
    print(f"stalled={stalled:.3f}s fixed={ok:.3f}s")
    assert ok < stalled  # => the offload restored concurrency
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

stalled=0.201s fixed=0.101s

import pydantic  # => to catch ValidationError
from pydantic import BaseModel, field_validator  # => co-12: field_validator is the custom-rule verb
 
 
class User(BaseModel):  # => a model with a custom email rule (co-12)
    email: str  # => a plain string at the type level
 
    @field_validator("email")  # => the custom rule runs during validation (co-13)
    @classmethod
    def must_have_at(cls, value: str) -> str:  # => receives the raw value
        if "@" not in value:  # => the rule
            raise ValueError("email must contain '@'")  # => a violation -> ValidationError/422 (co-13)
        return value  # => accepted value
 
 
def main() -> None:
    ok = User(email="a@b.com")  # => accepted
    print(ok.email)  # => Output: a@b.com
    try:
        User(email="bad")  # => no @ -> rejected (co-13)
        raised = False
    except pydantic.ValidationError:
        raised = True
    print(raised)  # => Output: True
    assert raised is True
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

a@b.com
True

import asyncio  # => co-23: Lock guards the critical section
 
 
class Counter:  # => a lock-guarded shared counter (co-23)
    def __init__(self) -> None:
        self._value = 0  # => the shared mutable state
        self._lock = asyncio.Lock()  # => only one coroutine in the critical section at a time
 
    async def inc(self) -> int:  # => a guarded read-modify-write
        async with self._lock:  # => atomic across coroutines (co-23)
            self._value += 1  # => the critical section
            return self._value  # => the new value
 
 
async def main() -> None:  # => N concurrent increments
    counter = Counter()  # => the shared counter
    n = 100  # => number of concurrent increments
    await asyncio.gather(*(counter.inc() for _ in range(n)))  # => all run concurrently (co-04)
    print(counter._value)  # => Output: 100 -- no lost updates
    assert counter._value == n  # => the lock made every increment count
 
 
if __name__ == "__main__":
    asyncio.run(main())

Run: python3 kata.py

Output:

100

from collections.abc import Iterator  # => the shape of a yield-provider (co-15)
 
 
class Conn:  # => a stand-in resource
    def __init__(self, name: str) -> None:
        self.name = name
        self.closed = False
 
    def close(self) -> None:  # => teardown
        self.closed = True
 
 
def run_with_dependency(  # => mimics what Depends does per request (co-15)
    provider: Iterator[Conn], handler,
) -> str:
    resource = next(provider)  # => resolve the resource (the part before yield)
    try:
        return handler(resource)  # => the handler just RECEIVES the resource
    finally:
        try:  # => always run teardown, even on error
            next(provider)  # => drive the provider past yield -> runs its teardown
        except StopIteration:  # => the generator finished -- expected after teardown
            pass
 
 
def get_conn() -> Iterator[Conn]:  # => a yield-provider (co-15)
    conn = Conn("primary")  # => setup
    yield conn  # => hand it out
    conn.close()  # => teardown -- runs after the handler returns/raises
 
 
def use_conn(c: Conn) -> str:  # => a handler that just uses the resource
    return f"used {c.name}"
 
 
def main() -> None:
    result = run_with_dependency(get_conn(), use_conn)  # => provider + handler
    print(result)  # => Output: used primary
    assert result == "used primary"
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

used primary

class NotFound(Exception):  # => a domain error (co-17)
    pass
 
 
class Conflict(Exception):  # => another domain error
    pass
 
 
def dispatch(handler, exc_map: dict[type[Exception], tuple[int, str]]):  # => runs handler, maps failures (co-17)
    try:
        return 200, handler()  # => the success path
    except Exception as exc:  # => a domain error raised by the handler
        for exc_type, mapping in exc_map.items():  # => find the mapped class
            if isinstance(exc, exc_type):
                return mapping  # => the mapped (status, body)
        raise  # => unmapped -> re-raise (co-17)
 
 
def read_missing():  # => raises NotFound
    raise NotFound()
 
 
def create_conflict():  # => raises Conflict
    raise Conflict()
 
 
def main() -> None:
    mapping = {NotFound: (404, "not found"), Conflict: (409, "conflict")}  # => the central mapping (co-17)
    print(dispatch(read_missing, mapping))  # => Output: (404, 'not found')
    print(dispatch(create_conflict, mapping))  # => Output: (409, 'conflict')
    assert dispatch(read_missing, mapping) == (404, "not found")
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

(404, 'not found')
(409, 'conflict')

from typing import TypedDict
 
 
class Row(TypedDict):  # => a typed row
    id: int
    status: str
 
 
class Page(TypedDict):  # => the paginated envelope (co-11)
    items: list[Row]
    total: int  # => the FILTERED count
    next: int | None  # => None at the end
 
 
def list_page(rows: list[Row], status: str | None, limit: int, offset: int) -> Page:  # => co-11
    filtered = [r for r in rows if status is None or r["status"] == status]  # => filter step
    total = len(filtered)  # => the filtered count, not the whole list (co-11)
    page = filtered[offset : offset + limit]  # => pagination step over the filtered set
    nxt = offset + limit  # => the next page's offset
    return {"items": page, "total": total, "next": nxt if nxt < total else None}  # => None at end
 
 
def main() -> None:
    rows: list[Row] = [  # => 4 rows: 2 done, 2 todo
        {"id": 1, "status": "done"},
        {"id": 2, "status": "todo"},
        {"id": 3, "status": "done"},
        {"id": 4, "status": "todo"},
    ]
    p = list_page(rows, status="done", limit=1, offset=0)  # => first page of done-only
    print([r["id"] for r in p["items"]], p["total"], p["next"])  # => [1] 2 1 -- total is the FILTERED count
    assert [r["id"] for r in p["items"]] == [1]
    assert p["total"] == 2  # => 2 done rows total, even though only 1 is on this page
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

[1] 2 1

ALLOWED_OUT_FIELDS = {"name"}  # => the output model's allowed fields (co-14)
 
 
def filter_output(input_dict: dict[str, object]) -> dict[str, object]:  # => mimics response_model filtering
    return {k: v for k, v in input_dict.items() if k in ALLOWED_OUT_FIELDS}  # => keep only allowed fields (co-14)
 
 
def main() -> None:
    internal = {"name": "widget", "secret": "cost-is-3.50"}  # => the internal object has a secret
    out = filter_output(internal)  # => the output model strips it
    print(out)  # => Output: {'name': 'widget'}
    assert out == {"name": "widget"}  # => secret never reaches the output (co-14)
    assert "secret" not in out
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

{'name': 'widget'}

import asyncio  # => co-06: asyncio.timeout enforces a deadline
 
 
async def call_with_retry(func, attempts: int, timeout: float):  # => bounded retry + per-call deadline
    last_error: Exception = RuntimeError("no attempts made")  # => remembers the last failure
    for _ in range(attempts):  # => a BOUNDED number of attempts (co-06)
        try:
            async with asyncio.timeout(timeout):  # => cut off a slow call after `timeout` (co-06)
                return await func()  # => success -> return immediately
        except (RuntimeError, TimeoutError) as exc:  # => transient failure or timeout -> retry
            last_error = exc  # => record and retry
    raise last_error  # => exhausted -> re-raise (a handler maps this to 503, co-17)
 
 
attempts_made = 0  # => a mutable counter for the mock
 
 
async def flaky():  # => fails twice, then succeeds
    global attempts_made
    attempts_made += 1
    await asyncio.sleep(0.01)  # => simulate latency (co-02)
    if attempts_made < 3:  # => fail the first two
        raise RuntimeError("transient")
    return "ok"  # => succeed on the third
 
 
async def main() -> None:
    result = await call_with_retry(flaky, attempts=5, timeout=1.0)  # => succeeds on attempt 3
    print(result)  # => Output: ok
    assert result == "ok"
 
 
if __name__ == "__main__":
    asyncio.run(main())

Run: python3 kata.py

Output:

ok

DEFAULTS: dict[str, object] = {"env": "dev", "port": 8000}  # => co-24: typed defaults
TYPES: dict[str, type] = {"env": str, "port": int}  # => the expected type per field
 
 
def load_settings(raw: dict[str, object]) -> dict[str, object]:  # => mimics pydantic-settings (co-24)
    out: dict[str, object] = dict(DEFAULTS)  # => start from defaults
    for key, value in raw.items():  # => apply overrides
        if key in TYPES and not isinstance(value, TYPES[key]):  # => wrong type -> fail fast (co-24)
            raise ValueError(f"{key} must be {TYPES[key].__name__}, got {type(value).__name__}")
        out[key] = value  # => accepted override
    return out  # => resolved config
 
 
def main() -> None:
    ok = load_settings({"port": 9000})  # => override applies, env stays default
    print(ok)  # => Output: {'env': 'dev', 'port': 9000}
    assert ok["port"] == 9000
    raised = False
    try:
        load_settings({"port": "not-an-int"})  # => wrong type -> ValueError (co-24)
    except ValueError:
        raised = True
    print(raised)  # => Output: True
    assert raised is True
 
 
if __name__ == "__main__":
    main()

Run: python3 kata.py

Output:

{'env': 'dev', 'port': 9000}
True

Because the two waits overlap on the single event loop: when the first coroutine awaits, the loop switches to the second, so both waits are in flight simultaneously. The loop interleaves the waits rather than running them one after another, so the total time is the max (~0.1s), not the sum. There is still only one thread -- concurrency is about interleaving waits, not parallel execution.

asyncio.sleep yields to the loop, letting other coroutines run during the wait; time.sleep blocks the thread, so the loop cannot switch to anything else for the whole 0.1s. On a busy server that means every concurrent request queues behind the blocking call. The two look identical in isolation and differ catastrophically under load.

Because response_model=ItemOut guarantees the secret never leaves the server: the output filter strips any field not on ItemOut, regardless of what the handler returns. A single model with an optional field relies on handler discipline (don't return the secret) -- one refactor that accidentally returns the whole object leaks it. Two models make the leak structurally impossible.

Because the synchronous driver blocks the thread on every query, stalling the event loop exactly like time.sleep (co-06). The handler is async def, but it awaits nothing useful during the query -- the loop is frozen. So every concurrent request queues behind each blocking query, and the async machinery buys nothing. The async driver is what closes the gap.

Because the dependency is an async generator: the yield hands the session to the handler, and the code after the yield (the await conn.close()) runs in the dependency's teardown phase, which FastAPI runs whether the handler returned normally or raised. The finally/post-yield cleanup is guaranteed, so the connection is never leaked.

Because FastAPI uses 422 Unprocessable Content (defined natively in RFC 9110) for request-validation failures specifically, distinguishing "the request was well-formed JSON but its content failed the declared schema" (422) from "the request was malformed" (400). The 422 carries a detail array naming each offending location, which is more useful to a client than a generic 400.

Because a pooled client reuses TCP connections across requests -- constructing one per request pays a fresh handshake (and a fresh pool) every call, throwing away the pooling benefit and increasing latency and resource use. The lifespan opens the pool once at startup and closes it once at shutdown, so every request shares it; per-request construction defeats the entire purpose of pooling.

Because the loop can switch coroutines between the read and the write: coroutine A reads count=5, then awaits; the loop runs coroutine B which also reads 5 and writes 6; A resumes and writes 6 -- one increment is lost. There are no OS threads, but there is interleaving at await points, and that interleaving is enough to lose updates. An asyncio.Lock makes the read-modify-write atomic across those interleavings.