Beginner Examples

"""Example 1: First Coroutine via asyncio Run."""
 
# => asyncio is the standard-library event loop -- no third-party package is needed (co-02)
import asyncio  # => the module that owns the loop, coroutines, Tasks, and gather
 
 
async def greet() -> str:  # => "async def" makes this a COROUTINE, not a normal function (co-01)
    # => calling greet() does NOT run the body -- it returns a coroutine OBJECT you must schedule
    await asyncio.sleep(0.01)  # => await SUSPENDS greet, handing control back to the loop for ~10ms
    # => while suspended the loop is free to run OTHER coroutines -- that is the whole point (co-02)
    return "hello from a coroutine"  # => the value the awaiter receives once this completes
 
 
def main() -> None:  # => a plain (non-async) entry point -- the bridge into the async world
    # => asyncio.run creates a FRESH event loop, runs one coroutine to completion, then closes it
    message = asyncio.run(greet())  # => the ONLY correct top-level driver for a coroutine (co-01)
    print(message)  # => Output: hello from a coroutine
 
 
if __name__ == "__main__":  # => only runs when executed directly, not on import
    main()  # => drives the coroutine via asyncio.run

hello from a coroutine

"""Example 2: Sequential Awaits Add Up."""
 
import asyncio  # => the event-loop module (co-02)
import time  # => only to MEASURE wall-clock time, never to "sleep" inside a coroutine (see ex-07)
 
 
async def pause(seconds: float) -> str:  # => a coroutine that simply waits, then returns a label
    # => await yields to the loop for the given duration -- other coroutines could run meanwhile (co-01)
    await asyncio.sleep(seconds)  # => the awaitable here is asyncio.sleep's coroutine
    return f"paused {seconds}s"  # => returned to whoever awaits this coroutine
 
 
async def main() -> tuple[str, str, float]:  # => two awaits IN SEQUENCE -- the second starts after the first ends
    start = time.perf_counter()  # => capture start time BEFORE the first await
    first = await pause(0.10)  # => awaited alone: the loop waits ~0.10s, doing nothing else here
    # => only AFTER first completes does the second await even begin -- they are SERIAL
    second = await pause(0.10)  # => a second ~0.10s wait, stacked after the first
    elapsed = time.perf_counter() - start  # => total wall-clock cost of BOTH waits
    # => elapsed is ~0.20s -- the SUM, because nothing ran concurrently (contrast ex-03)
    return first, second, elapsed
 
 
if __name__ == "__main__":  # => only runs when executed directly
    f, s, elapsed = asyncio.run(main())  # => drive the async main with a fresh loop
    print(f)  # => Output: paused 0.1s
    print(s)  # => Output: paused 0.1s
    print(f"elapsed={elapsed:.3f}s")  # => Output: elapsed=0.20Xs (the sum of both waits)

paused 0.1s
paused 0.1s
elapsed=0.200s

"""Example 3: Concurrent Awaits with gather."""
 
import asyncio  # => the event-loop module -- gather lives here (co-04)
import time  # => wall-clock measurement only
 
 
async def pause(seconds: float) -> str:  # => the SAME coroutine as ex-02
    await asyncio.sleep(seconds)  # => yields to the loop for the given duration (co-01)
    return f"paused {seconds}s"  # => labelled result
 
 
async def main() -> tuple[list[str], float]:  # => results come back IN ORDER, time is the MAX
    start = time.perf_counter()  # => capture start before fanning out
    # => asyncio.gather SCHEDULES both coroutines concurrently and waits for ALL of them (co-04)
    # => the loop interleaves the two waits, so they overlap instead of stacking
    results = await asyncio.gather(pause(0.10), pause(0.10))  # => returns a list, order preserved
    elapsed = time.perf_counter() - start  # => total cost is ~the SLOWER call, not the sum
    # => contrast ex-02's ~0.20s: same two waits, but ~0.10s here -- the concurrency win (co-04)
    return results, elapsed
 
 
if __name__ == "__main__":  # => only runs when executed directly
    results, elapsed = asyncio.run(main())  # => drive the async main
    print(results)  # => Output: ['paused 0.1s', 'paused 0.1s']
    print(f"elapsed={elapsed:.3f}s")  # => Output: elapsed=0.10Xs (the max, not the sum)

['paused 0.1s', 'paused 0.1s']
elapsed=0.101s

"""Example 4: Scheduling Background Work as a Task."""
 
import asyncio  # => create_task and gather live here (co-04)
 
 
async def step(label: str, seconds: float) -> str:  # => a unit of background work
    await asyncio.sleep(seconds)  # => yields to the loop while "working" (co-01)
    return f"{label} done"  # => the value the Task resolves to
 
 
async def main() -> list[str]:  # => schedules work, does other things, then collects results
    # => create_task WRAPS a coroutine in a Task and schedules it on the loop IMMEDIATELY (co-04)
    # => the coroutine starts running as soon as the loop gets control -- it does not wait to be awaited
    task = asyncio.create_task(step("background", 0.05))  # => running concurrently, in the background
    # => this await runs CONCURRENTLY with the task above -- both make progress on one loop
    foreground = await step("foreground", 0.05)  # => awaited directly, in the foreground
    # => awaiting the Task COLLECTS its result; if it already finished, this returns immediately
    background = await task  # => no extra wait if the task completed during the foreground await
    return [foreground, background]  # => both completed, in the order we collect them
 
 
if __name__ == "__main__":  # => only runs when executed directly
    results = asyncio.run(main())  # => drive the async main
    print(results)  # => Output: ['foreground done', 'background done']

['foreground done', 'background done']

"""Example 5: Managing a Resource with async with."""
 
import asyncio  # => the event-loop module (co-02)
 
 
class AsyncConnection:  # => models a resource whose acquire/release themselves need to await (co-05)
    async def __aenter__(self) -> "AsyncConnection":  # => async context-manager ENTRY -- runs on "async with"
        await asyncio.sleep(0.01)  # => simulate an async "connect" that yields to the loop
        print("opened")  # => confirms setup ran
        return self  # => the bound resource handed to the "as" name
 
    async def __aexit__(self, exc_type: object, exc: object, tb: object) -> None:  # => EXIT -- runs on block exit
        await asyncio.sleep(0.01)  # => simulate an async "close" that yields to the loop
        print("closed")  # => confirms teardown ran even if the body raised
 
 
async def main() -> None:  # => demonstrates the setup/body/teardown order
    async with AsyncConnection() as conn:  # => __aenter__ runs, then the body, then __aexit__ (co-05)
        _ = conn  # => the live resource is available only INSIDE the block
        print("using")  # => body runs between "opened" and "closed"
 
 
if __name__ == "__main__":  # => only runs when executed directly
    asyncio.run(main())  # => Output order: opened / using / closed

opened
using
closed

"""Example 6: Consuming an Async Stream with async for."""
 
import asyncio  # => the event-loop module (co-02)
from collections.abc import AsyncIterator  # => the typed return shape of an async generator (co-05)
 
 
async def emit(count: int) -> AsyncIterator[int]:  # => an ASYNC GENERATOR -- "async def" + "yield"
    for i in range(count):  # => produces one item per iteration
        await asyncio.sleep(0.01)  # => yields to the loop between items (co-01)
        yield i  # => hands one value to the consumer, then resumes here on the next pull (co-05)
 
 
async def main() -> list[int]:  # => consumes the whole stream into a list
    collected: list[int] = []  # => accumulator for the values pulled from the stream
    # => "async for" PULLS the next item, awaiting the generator between each one (co-05)
    async for value in emit(3):  # => each iteration awaits the generator's next "yield"
        collected.append(value)  # => values arrive one at a time, 0 then 1 then 2
    return collected  # => the full stream, materialised in order
 
 
if __name__ == "__main__":  # => only runs when executed directly
    result = asyncio.run(main())  # => drive the async main
    print(result)  # => Output: [0, 1, 2]

[0, 1, 2]

"""Example 7: A Blocking Call Stalls the Loop.
 
The single most common async bug: a SYNCHRONOUS blocking call inside a coroutine freezes the WHOLE loop,
not just that coroutine. This example proves it, then shows the two correct fixes.
"""
 
import asyncio  # => the event-loop module (co-02)
import time  # => time.sleep is BLOCKING -- the hazard itself (co-06)
 
 
async def good_wait(seconds: float) -> str:  # => the CORRECT async sleep -- yields to the loop
    await asyncio.sleep(seconds)  # => other coroutines keep running during this wait (co-01)
    return "good"
 
 
async def bad_wait(seconds: float) -> str:  # => the HAZARD: time.sleep blocks the entire loop (co-06)
    time.sleep(seconds)  # => NO await -- the loop CANNOT switch to any other coroutine while this runs
    return "bad"
 
 
async def fixed_wait(seconds: float) -> str:  # => the FIX for genuinely blocking code: offload it (co-06, ex-08)
    loop = asyncio.get_running_loop()  # => fetch the loop this coroutine is running on
    # => run_in_executor pushes the blocking call onto a thread, so the loop stays responsive
    await loop.run_in_executor(None, time.sleep, seconds)  # => the loop is free while the thread blocks
    return "fixed"
 
 
async def main() -> dict[str, float]:  # => times BOTH orderings to expose the stall
    start = time.perf_counter()  # => baseline
    await asyncio.gather(bad_wait(0.10), good_wait(0.10))  # => bad blocks good: total ~0.20s, not ~0.10s
    stalled = time.perf_counter() - start  # => the stall made the concurrent pair take the SUM
    # => the fix: offloaded blocking work no longer blocks the loop, so good_wait overlaps it again
    start2 = time.perf_counter()  # => baseline for the fixed run
    await asyncio.gather(fixed_wait(0.10), good_wait(0.10))  # => both run concurrently again
    fixed = time.perf_counter() - start2  # => ~0.10s -- the max, concurrency restored
    return {"stalled": stalled, "fixed": fixed}
 
 
if __name__ == "__main__":  # => only runs when executed directly
    times = asyncio.run(main())  # => drive the async main
    print(f"stalled={times['stalled']:.3f}s")  # => Output: stalled=0.20Xs (blocking froze the loop)
    print(f"fixed={times['fixed']:.3f}s")  # => Output: fixed=0.10Xs (offload restored concurrency)

stalled=0.201s
fixed=0.101s

"""Example 8: Offloading Blocking Work to an Executor."""
 
import asyncio  # => the event-loop module (co-02)
import time  # => only to simulate + measure
 
 
def blocking_cpu_or_io(seconds: float) -> int:  # => a PLAIN synchronous function -- NOT a coroutine
    # => imagine a blocking DB driver, a CPU-heavy computation, or a C-extension call (co-06)
    time.sleep(seconds)  # => blocks the THREAD it runs on -- fine on a pool thread, fatal on the loop
    return 42  # => a plain return value, handed back across the executor boundary
 
 
async def main() -> tuple[int, float]:  # => keeps the loop responsive while a thread does the blocking work
    loop = asyncio.get_running_loop()  # => the loop this coroutine is running on
    start = time.perf_counter()  # => baseline
    # => run_in_executor schedules the blocking function on a thread pool, returning an awaitable (co-03, co-06)
    # => the FIRST arg is the executor (None = the default ThreadPoolExecutor); the loop stays free meanwhile
    result = await loop.run_in_executor(None, blocking_cpu_or_io, 0.10)  # => awaited -> resolves to the int
    # => a SECOND coroutine scheduled alongside would keep running during that 0.10s (co-02) -- the win
    elapsed = time.perf_counter() - start  # => ~0.10s on the thread, with the loop still responsive
    return result, elapsed
 
 
if __name__ == "__main__":  # => only runs when executed directly
    result, elapsed = asyncio.run(main())  # => drive the async main
    print(result)  # => Output: 42
    print(f"elapsed={elapsed:.3f}s")  # => Output: elapsed=0.10Xs (loop never stalled)

42
elapsed=0.101s

"""Example 9: Creating a Locked Environment with uv.
 
The uv workflow this example assumes:
  uv venv && source .venv/bin/activate
  uv pip install fastapi==0.139.0 pydantic==2.11.0
  uv run python example.py
 
This module confirms the locked install by reading the pinned versions back -- a reproducible environment
is what makes every later example see the same code. (co-07)
"""
 
import fastapi  # => the web framework pinned by the uv install above (co-07)
import pydantic  # => the validation library pinned alongside it (v2 line)
 
 
def installed_versions() -> tuple[str, str]:  # => reads installed metadata, no network call
    # => __version__ is plain module metadata pip/uv wrote at install time -- instant and offline
    return fastapi.__version__, pydantic.VERSION  # => a tuple of the two pinned version strings
 
 
if __name__ == "__main__":  # => run as: uv run python example.py
    fastapi_v, pydantic_v = installed_versions()  # => unpack the tuple
    print(f"fastapi=={fastapi_v}")  # => Output: fastapi==0.139.0 (the pinned line)
    print(f"pydantic=={pydantic_v}")  # => Output: pydantic==2.11.X (the v2 production line)

fastapi==0.139.0
pydantic==2.11.0

"""Example 10: Linting and Formatting Clean with ruff.
 
This module IS the lint-clean artifact. Run: ruff check example.py && ruff format --check example.py
-> zero findings. (co-08)
"""
 
from collections.abc import Sequence  # => the precise typed alias for "a sequence of ints"
 
 
def double_each(values: Sequence[int]) -> list[int]:  # => fully typed signature -- ruff and pyright both like it
    # => a list comprehension is idiomatic, short, and lint-clean -- no manual loop index needed (co-08)
    return [value * 2 for value in values]  # => one transformed value per input, in order
 
 
if __name__ == "__main__":  # => run directly to confirm behaviour
    result = double_each([1, 2, 3])  # => transform a small sequence
    print(result)  # => Output: [2, 4, 6]

All checks passed!

[2, 4, 6]

"""Example 11: Type Checking Clean with pyright.
 
This module is fully type-annotated and pyright-clean (strict). Run: pyright example.py -> 0 errors. (co-09)
"""
 
 
def parse_port(raw: str) -> int:  # => every parameter AND the return are annotated (co-09)
    # => a ValueError narrows "bad input" to a precise, typed failure instead of an untyped crash
    value = int(raw)  # => may raise ValueError -- pyright knows value is int here
    if not (1 <= value <= 65535):  # => a valid TCP/UDP port range check
        raise ValueError(f"port out of range: {value}")  # => typed-domain rejection (no bare string fallback)
    return value  # => pyright confirms an int is returned on every path
 
 
def build_url(host: str, port_raw: str) -> str:  # => composes the typed parser into a larger function
    port = parse_port(port_raw)  # => pyright knows port is int -- no cast needed anywhere
    return f"http://{host}:{port}"  # => a fully-typed string result
 
 
if __name__ == "__main__":  # => run directly to confirm behaviour
    print(build_url("localhost", "8000"))  # => Output: http://localhost:8000

0 errors, 0 warnings, 0 informations

http://localhost:8000

"""Example 12: A First FastAPI Route.
 
Run: uvicorn app:app --port 8000, then: curl localhost:8000/  (co-10)
"""
 
from fastapi import FastAPI  # => the web framework whose routing this example exercises (co-10)
 
# => the module-level name "app" is exactly what "uvicorn app:app" imports (module app, attribute app)
app = FastAPI()  # => an ASGI application object the server serves
 
 
@app.get("/")  # => decorator-based ROUTING: GET "/" maps to read_root (co-10, co-11)
def read_root() -> dict[str, str]:  # => returning a dict is auto-serialized to a JSON response body (co-14)
    # => no manual json.dumps, no manual Content-Type, no manual status line -- the framework does all three
    return {"msg": "hello"}  # => FastAPI sets Content-Type: application/json and status 200 automatically

{"msg":"hello"}

"""Example 13: A Typed Path Parameter.
 
Run: uvicorn app:app --port 8000, then: curl localhost:8000/items/5  (co-11)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.get("/items/{item_id}")  # => {item_id} is a path-parameter PLACEHOLDER in the route template (co-11)
def read_item(item_id: int) -> dict[str, int]:  # => the matching name + int type = a typed PATH param
    # => a NON-numeric segment (e.g. /items/abc) fails validation with a 422 before this body runs (co-13)
    return {"item_id": item_id}  # => the SAME typed int, echoed back as JSON (co-14)

{"item_id":5}

"""Example 14: A Typed Query Parameter.
 
Run: uvicorn app:app --port 8000, then: curl 'localhost:8000/items' and
curl 'localhost:8000/items?limit=3'  (co-11)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
@app.get("/items")  # => no "{limit}" placeholder here, so limit is inferred as a QUERY param (co-11)
def list_items(limit: int = 10) -> dict[str, int]:  # => a DEFAULT makes the param OPTIONAL ("= 10")
    # => "?limit=3" sets limit=3; omitting "limit" entirely uses the default 10 (co-11)
    return {"limit": limit}  # => echoes whichever value was actually used (co-14)

{"limit":10}
{"limit":3}

"""Example 15: A Pydantic Model as a Request Body.
 
Run: uvicorn app:app --port 8000, then:
curl -X POST -H 'Content-Type: application/json' -d '{"name":"widget","price":9.99}' localhost:8000/items
(co-12, co-11)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel  # => Pydantic models are FastAPI's validation vocabulary (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class Item(BaseModel):  # => a typed REQUEST-BODY model -- Pydantic validates incoming JSON against it (co-12)
    name: str  # => required string field -- a missing/wrong type fails validation (co-13)
    price: float  # => required float field -- a second, independent constraint
 
 
@app.post("/items")  # => POST is the method this route pairs a body with (co-11)
def create_item(item: Item) -> Item:  # => a Pydantic-model PARAM tells FastAPI to parse the BODY as JSON
    # => by the time this line runs, the body was ALREADY parsed + validated -- a bad body never reaches here
    return item  # => Pydantic re-serializes the model back to JSON on the way out (co-14)

{"name":"widget","price":9.99}

"""Example 16: Invalid Input Returns a 422.
 
The SAME app as ex-15 -- this example exists to POST an INVALID body and observe the automatic 422.
Run: uvicorn app:app --port 8000, then:
curl -s -w '\\nHTTP %{http_code}\\n' -X POST -H 'Content-Type: application/json' -d '{"name":"widget"}' localhost:8000/items
(co-13)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel  # => Pydantic models are the validation vocabulary (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class Item(BaseModel):  # => the shape a valid POST /items body must satisfy (co-12)
    name: str  # => required
    price: float  # => required -- omitting it is exactly the 422 this example triggers
 
 
@app.post("/items", status_code=201)  # => status 201 for a successful create (co-17)
def create_item(item: Item) -> Item:  # => validation runs BEFORE this body -- bad input never reaches here
    # => sending {"name":"widget"} (no price) fails validation and returns a 422, never reaching this return
    return item  # => only a fully-valid body reaches this line

{"detail":[{"type":"missing","loc":["body","price"],"msg":"Field required","input":{"name":"widget"}}]}
HTTP 422

"""Example 17: Shaping the Output with a response model.
 
Run: uvicorn app:app --port 8000, then:
curl -X POST -H 'Content-Type: application/json' -d '{"name":"widget","secret_note":"internal"}' localhost:8000/items
(co-14)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
app = FastAPI()  # => the ASGI application uvicorn serves
 
 
class ItemIn(BaseModel):  # => the INPUT shape -- includes a field the response must NEVER leak (co-13)
    name: str  # => safe to echo back
    secret_note: str  # => sent by the client, but the response model below omits it
 
 
class ItemOut(BaseModel):  # => the OUTPUT shape -- a strict subset of ItemIn's fields (co-14)
    name: str  # => the ONLY field this model is allowed to expose
 
 
@app.post("/items", response_model=ItemOut)  # => response_model FILTERS the return value to ItemOut's fields
def create_item(item: ItemIn) -> ItemOut:  # => validated by ItemIn on the way in, filtered by ItemOut out
    # => even if the handler leaked item.secret_note, response_model would strip it -- filtering is on the
    # => OUTGOING side, independent of what the handler body computes (co-14)
    return ItemOut(name=item.name)  # => the response body contains name ONLY, never secret_note

{"name":"widget"}

"""Example 18: OpenAPI Docs Are Generated for Free.
 
FastAPI derives the full OpenAPI schema from the typed routes -- every path, parameter, model, and status
appears with no extra authoring. Run as a server: uvicorn app:app --port 8000, then curl localhost:8000/openapi.json
-- or run directly to print the generated schema's paths. (co-20)
"""
 
from fastapi import FastAPI  # => the web framework (co-10)
from pydantic import BaseModel  # => Pydantic models (co-12)
 
app = FastAPI(title="OpenAPI Demo")  # => the title appears in the generated schema's info block (co-20)
 
 
class Item(BaseModel):  # => this model becomes a schema COMPONENT in the generated OpenAPI doc (co-20)
    name: str  # => required string
    price: float  # => required float
 
 
@app.get("/items/{item_id}")  # => becomes a paths entry: GET /items/{item_id} (co-20)
def read_item(item_id: int) -> dict[str, int]:  # => item_id shows up as an integer path PARAMETER (co-11)
    return {"item_id": item_id}  # => the return type informs the response schema (co-14)
 
 
@app.post("/items")  # => becomes a paths entry: POST /items, body referencing the Item schema (co-20)
def create_item(item: Item) -> Item:  # => the Item body and Item response both appear in the schema (co-12)
    return item  # => round-tripped
 
 
if __name__ == "__main__":  # => run directly to INSPECT the generated schema without starting a server
    schema = app.openapi()  # => returns the dict FastAPI would serve at /openapi.json (co-20)
    print(sorted(schema["paths"].keys()))  # => Output: ['/items', '/items/{item_id}']
    print(schema["components"]["schemas"]["Item"]["required"])  # => Output: ['name', 'price']

['/items', '/items/{item_id}']
['name', 'price']