Foundations Updated 2026-09 View as Markdown

Async and Await

Concurrency for I/O-bound work — and the one mistake that silently makes an async program slower than the synchronous version.

Async is for waiting on many things at once — network calls, database queries, file I/O. It does not make computation faster; it stops you idling while the network responds.

import asyncio

async def fetch(name: str, delay: float) -> str:
    await asyncio.sleep(delay)         # yields control while waiting
    return f"{name} done"

async def main() -> None:
    result = await fetch("first", 1.0)
    print(result)

asyncio.run(main())

async def creates a coroutine. Calling it does nothing on its own — it returns a coroutine object that must be awaited or scheduled:

fetch("a", 1)            # RuntimeWarning: coroutine was never awaited
await fetch("a", 1)      # runs it

Concurrency is the point#

async def main() -> None:
    a = await fetch("first", 1.0)      # 1 second
    b = await fetch("second", 1.0)     # then another second
    # total: 2 seconds — this is just slow synchronous code

Sequential awaits buy you nothing. Run them together:

async def main() -> None:
    a, b = await asyncio.gather(
        fetch("first", 1.0),
        fetch("second", 1.0),
    )
    # total: 1 second

Consecutive await lines with no data dependency between them are the signal that something should be a gather.

TaskGroup#

The modern replacement for gather, and the better default:

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        first = tg.create_task(fetch("first", 1.0))
        second = tg.create_task(fetch("second", 1.0))
    # both are complete here
    print(first.result(), second.result())

The advantage over gather is failure handling: if one task raises, the others are cancelled and the errors are collected into an ExceptionGroup. With gather, a failure leaves the siblings running in the background, still consuming resources.

try:
    async with asyncio.TaskGroup() as tg:
        ...
except* ValueError as eg:              # except* handles ExceptionGroup
    for err in eg.exceptions:
        print("bad value:", err)

The mistake that matters#

result = await asyncio.to_thread(expensive_sync_function, arg)

to_thread runs blocking code in a worker thread so the loop keeps running. It is the escape hatch for a library with no async version.

Catch it mechanically: turn on ruff's ASYNC rules, and run tests with PYTHONASYNCIODEBUG=1, which warns when a coroutine blocks for more than 100ms. See the failure-mode catalogue.

Bounding concurrency#

gather over a large list starts everything at once — which means a rate-limited API returns 429s, or you exhaust file descriptors:

async def fetch_all(ids: list[str]) -> list[str]:
    sem = asyncio.Semaphore(10)          # at most 10 in flight

    async def one(uid: str) -> str:
        async with sem:
            return await fetch(uid, 0.1)

    return await asyncio.gather(*(one(i) for i in ids))

When the list size comes from data rather than from your source code, bound it.

Timeouts and cancellation#

try:
    async with asyncio.timeout(5.0):
        data = await slow_operation()
except TimeoutError:
    print("gave up after 5 seconds")

Cancellation is cooperative: asyncio raises CancelledError at the next await point. Clean up with try/finally, and never swallow it:

async def worker() -> None:
    try:
        while True:
            await do_work()
    except asyncio.CancelledError:
        await flush()      # clean up
        raise              # ALWAYS re-raise — swallowing breaks shutdown
    finally:
        await close()

Async iterators and context managers#

async with httpx.AsyncClient() as client:      # __aenter__ / __aexit__
    r = await client.get(url)

async for row in cursor:                        # __aiter__ / __anext__
    process(row)

async def paginate(url: str):
    while url:
        page = await fetch_page(url)
        for item in page["items"]:
            yield item                          # an async generator
        url = page.get("next")

async for item in paginate("/api/items"):
    print(item)

When not to use async#

Async adds real complexity — a coloured-function split where async code can only be called from async code, harder debugging, and a whole category of new bugs.

Do not use it for: CPU-bound work (use multiprocessing), scripts making a handful of sequential calls, or code where the concurrency would be one or two operations. Threads are simpler and often sufficient.

Do use it for: servers handling many simultaneous connections, clients making hundreds of requests, and anything where you would otherwise be waiting on I/O most of the time.

Exercise#

import asyncio

async def fetch_price(sku: str) -> int:
    await asyncio.sleep(0.1)
    return len(sku) * 100

async def main() -> None:
    skus = ["A1", "B22", "C333", "D4444"]
    # 1. Fetch all prices CONCURRENTLY and print the total.
    # 2. Limit concurrency to 2 with a Semaphore.
    # 3. Wrap the whole thing in a 1-second timeout.
    print(skus)

asyncio.run(main())

Common questions#

Does async make my code faster?#

Only for I/O-bound work, and only when operations actually overlap. It removes waiting, not computation — a CPU-heavy function is exactly as slow inside a coroutine, and it blocks everything else while it runs.

Why is my async code no faster than the sync version?#

The two usual causes: you are awaiting sequentially instead of using gather or a TaskGroup, or something in the path is a blocking call that freezes the loop. Check for requests, time.sleep or a synchronous database driver first.

gather or TaskGroup?#

TaskGroup for new code. When one task fails it cancels the siblings and reports errors as a group, whereas gather leaves the others running in the background. Use gather when you genuinely want independent results and return_exceptions=True.