Asyncio Patterns Catalog

Every core asyncio pattern with runnable code. Examples use small fakes (asyncio.sleep for latency) so they run offline; swap in your real LLM, embedding or vector-store client. Baseline is Python 3.9; newer features are tagged.

Mental model
Coroutines, tasks and futures
TaskGroup (3.11+)
gather
TaskGroup emulation for 3.9 and 3.10
wait and as_completed
Hedged requests and racing models
Worker pools on a queue
Multi-stage pipelines
Async generators
Async context managers
Calling blocking code
Threads calling into the loop
Background tasks
Eager tasks (3.12+)
Synchronization primitives
Event loop lifecycle
Context variables

Mental model

One thread runs the event loop. The loop runs one callback or task step at a time; a task step runs until it hits an await on something that is not ready, then the loop switches to another ready task. Consequences:

Code between two awaits is atomic with respect to other tasks (no lock needed for a dict update with no await in between).
Anything that does not yield (a blocking call, a CPU loop) stops every other task.
Concurrency comes from tasks, not from await. await a(); await b() is sequential.

Coroutines, tasks and futures

python

import asyncio
import time


async def call_model(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return name


async def main() -> None:
    start = time.perf_counter()
    await call_model("a", 0.1)                 # sequential: 0.2 s total
    await call_model("b", 0.1)
    seq = time.perf_counter() - start

    start = time.perf_counter()
    ta = asyncio.create_task(call_model("a", 0.1))   # both scheduled now
    tb = asyncio.create_task(call_model("b", 0.1))
    print(await ta, await tb)
    conc = time.perf_counter() - start
    print(f"sequential ~{seq:.1f}s, concurrent ~{conc:.1f}s")


asyncio.run(main())
# Output:
# a b
# sequential ~0.2s, concurrent ~0.1s

	Coroutine	Task	Future
What	paused function	coroutine scheduled on the loop	placeholder for a result
Starts	when awaited	at the next loop iteration (eager: immediately, 3.12)	n/a
Cancel	n/a (close())	`task.cancel()`	`fut.cancel()`
Result	value of `await`	`await task` / `task.result()`	`await fut` / `fut.result()`
Await twice	RuntimeError	fine, same result	fine

asyncio.ensure_future is legacy: it wraps coroutines in tasks and passes futures through. Use create_task (or a TaskGroup) for coroutines.

TaskGroup (3.11+)

Structured concurrency: tasks cannot outlive the async with block. When any task fails, the group cancels the others, waits for them, then raises an ExceptionGroup of every non-cancellation error.

python

import asyncio


async def embed(doc: str) -> list[float]:
    await asyncio.sleep(0.01)
    if doc == "corrupt":
        raise ValueError(f"cannot embed {doc!r}")
    return [float(len(doc))]


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(embed(d), name=f"embed-{d}") for d in ["alpha", "corrupt", "gamma"]]
    except* ValueError as eg:
        print("failed:", [str(e) for e in eg.exceptions])
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(embed(d)) for d in ["alpha", "beta"]]
    print([t.result() for t in tasks])


asyncio.run(main())
# Output:
# failed: ["cannot embed 'corrupt'"]
# [[5.0], [4.0]]

Rules:

Read results with t.result() after the block; inside the block tasks may still be running.
The body of the async with is itself part of the group: if it raises, the tasks are cancelled.
Tasks added from inside a running child (tg.create_task from a task) are fine while the group is open.
KeyboardInterrupt/SystemExit in a child propagate as themselves, not wrapped.
To keep going after failures, catch inside the child (try/except in a wrapper coroutine) and return an error value; the group only cancels on exceptions that escape a child.

gather

python

results = await asyncio.gather(*coros)                           # first error propagates, others keep running
results = await asyncio.gather(*coros, return_exceptions=True)   # errors returned in place

Semantics people get wrong:

Without return_exceptions, the first exception is raised to the awaiter, but the other awaitables are not cancelled. If the caller then returns, they keep running unowned.
If the gather itself is cancelled, all children are cancelled.
With return_exceptions=True, CancelledError of a child is also returned as a value (it is a BaseException), so filter with isinstance(r, BaseException), not Exception.
Results are in argument order regardless of completion order.
Passing an empty list returns [] immediately.

TaskGroup emulation for 3.9 and 3.10

python

import asyncio
from typing import Awaitable, List, TypeVar

T = TypeVar("T")


async def gather_all_or_cancel(*aws: Awaitable[T]) -> List[T]:
    """Like TaskGroup: on the first failure cancel the rest, wait for them, re-raise."""
    tasks = [asyncio.ensure_future(a) for a in aws]
    try:
        return await asyncio.gather(*tasks)
    except BaseException:
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)   # reap; never leave orphans
        raise


async def ok(x: int) -> int:
    await asyncio.sleep(0.05)
    return x


async def boom() -> int:
    await asyncio.sleep(0.01)
    raise RuntimeError("upstream 500")


async def main() -> None:
    slow = ok(1)
    try:
        await gather_all_or_cancel(slow, boom())
    except RuntimeError as err:
        print("raised:", err)
    print(await gather_all_or_cancel(ok(1), ok(2)))


asyncio.run(main())
# Output:
# raised: upstream 500
# [1, 2]

The anyio library offers task groups with the same semantics on 3.9+ (anyio.create_task_group()), and exceptiongroup backports ExceptionGroup/catch() for pre-3.11 code.

wait and as_completed

asyncio.wait(tasks, timeout=None, return_when=ALL_COMPLETED) returns (done, pending) sets and never raises task exceptions; you inspect them. Since 3.11 it accepts only tasks/futures, not bare coroutines.

python

import asyncio


async def score(i: int) -> int:
    await asyncio.sleep(0.02 * (5 - i))     # later items finish first
    return i


async def main() -> None:
    tasks = [asyncio.create_task(score(i)) for i in range(5)]
    order = []
    for fut in asyncio.as_completed(tasks):  # 3.13+: `async for task in asyncio.as_completed(tasks)`
        order.append(await fut)
    print("completion order:", order)

    tasks = [asyncio.create_task(score(i)) for i in range(5)]
    done, pending = await asyncio.wait(tasks, timeout=0.05)
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)
    print("done in time:", sorted(t.result() for t in done))


asyncio.run(main())
# Output:
# completion order: [4, 3, 2, 1, 0]
# done in time: [3, 4]

Use as_completed for progress reporting and "stop after the first K good results"; remember to cancel what is left when you stop early. wait(timeout=...) does not cancel pending tasks for you.

Hedged requests and racing models

Send the same request to a second replica (or a fallback model) if the first is slow, keep whichever answers first, cancel the loser.

python

import asyncio


async def replica(name: str, delay: float) -> str:
    await asyncio.sleep(delay)
    return f"answer from {name}"


async def hedged(hedge_after: float = 0.05) -> str:
    primary = asyncio.create_task(replica("primary", 0.2))
    done, _ = await asyncio.wait({primary}, timeout=hedge_after)
    if done:
        return primary.result()
    backup = asyncio.create_task(replica("backup", 0.02))
    done, pending = await asyncio.wait({primary, backup}, return_when=asyncio.FIRST_COMPLETED)
    for t in pending:
        t.cancel()
    await asyncio.gather(*pending, return_exceptions=True)
    winner = done.pop()
    return winner.result()          # re-raises if the winner failed; add fallback logic as needed


print(asyncio.run(hedged()))
# Output: answer from backup

Hedging doubles cost for the hedged fraction; set hedge_after near your p95 latency so only ~5% of requests are duplicated, and never hedge non-idempotent calls (tool calls with side effects, writes).

Worker pools on a queue

A fixed number of workers pulling from a bounded queue: constant memory, natural backpressure, easy per-item error handling. Preferred over "one task per item" when the input is large or unbounded (a corpus, a Kafka topic, a crawl frontier).

python

import asyncio


async def process(doc_id: int) -> None:
    await asyncio.sleep(0.001)
    if doc_id == 7:
        raise ValueError("unparseable PDF")


async def run(doc_ids: list[int], n_workers: int = 4) -> tuple[int, list[int]]:
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=n_workers * 2)
    ok, failed = 0, []

    async def worker() -> None:
        nonlocal ok
        while True:
            doc_id = await queue.get()
            try:
                await process(doc_id)
                ok += 1
            except Exception:
                failed.append(doc_id)
            finally:
                queue.task_done()

    workers = [asyncio.create_task(worker(), name=f"worker-{i}") for i in range(n_workers)]
    try:
        for d in doc_ids:
            await queue.put(d)
        await queue.join()
    finally:
        for w in workers:
            w.cancel()
        await asyncio.gather(*workers, return_exceptions=True)
    return ok, failed


print(asyncio.run(run(list(range(20)))))
# Output: (19, [7])

Details:

task_done() must be called exactly once per get(), in finally, or join() hangs forever.
If a worker can die (an exception escapes), join() may hang; the try/except inside the loop prevents it.
A worker that crashes inside a TaskGroup cancels the whole pool; that is often what you want for all-or-nothing batch jobs.
3.13+: queue.shutdown() makes blocked get() calls raise asyncio.QueueShutDown (after the queue drains, unless immediate=True), which is a cleaner stop signal than cancelling workers. Before 3.13 use a sentinel (None) per worker or cancellation.

Multi-stage pipelines

Separate stages (fetch -> parse -> embed -> upsert) with bounded queues, so each stage gets its own concurrency and the slowest stage throttles the rest.

python

import asyncio

STOP = object()


def stage(name, inq, outq, fn, workers):
    async def run():
        while True:
            item = await inq.get()
            try:
                if item is STOP:
                    return
                result = await fn(item)
                if outq is not None:
                    await outq.put(result)
            finally:
                inq.task_done()

    tasks = [asyncio.create_task(run(), name=f"{name}-{i}") for i in range(workers)]
    return tasks


async def fetch(url):
    await asyncio.sleep(0.001)
    return f"<html>{url}</html>"


async def parse(html):
    return html.removeprefix("<html>").removesuffix("</html>")


async def embed(text):
    await asyncio.sleep(0.001)
    return (text, len(text))


async def main():
    q1, q2, q3 = (asyncio.Queue(maxsize=8) for _ in range(3))
    results = []

    async def sink(item):
        results.append(item)

    fetchers = stage("fetch", q1, q2, fetch, 4)
    parsers = stage("parse", q2, q3, parse, 2)
    embedders = stage("embed", q3, None, lambda t: _embed_and_sink(t, sink), 3)
    for i in range(10):
        await q1.put(f"doc{i}")
    for group, q in ((fetchers, q1), (parsers, q2), (embedders, q3)):
        for _ in group:
            await q.put(STOP)          # one stop marker per worker, after the real items
        await asyncio.gather(*group)   # stage drained before stopping the next one
    print(len(results), sorted(results)[:2])


async def _embed_and_sink(text, sink):
    await sink(await embed(text))


asyncio.run(main())
# Output: 10 [('doc0', 4), ('doc1', 4)]

Shutting stages down in order (stop markers after the data, gather each stage before stopping the next) guarantees nothing is dropped. For CPU-heavy stages (PDF parsing) call a process pool inside fn.

Async generators

python

import asyncio
from contextlib import aclosing          # 3.10+


async def token_stream(prompt: str):
    try:
        for tok in prompt.split():
            await asyncio.sleep(0)
            yield tok
    finally:
        print("stream closed")           # runs on exhaustion, break (with aclosing) and cancellation


async def main() -> None:
    async with aclosing(token_stream("the quick brown fox")) as stream:
        async for tok in stream:
            if tok == "brown":
                break                    # aclosing() runs the generator's finally right here
    print("after loop")


asyncio.run(main())
# Output:
# stream closed
# after loop

Without aclosing, breaking out of async for leaves the generator suspended until it is garbage-collected or the loop shuts down (asyncio.run finalizes async generators at exit), so cleanup is delayed.
Never yield inside an async with asyncio.timeout() or TaskGroup block in a generator that the consumer may stop early: the cancel scope would be exited in a different task context. Collect into a queue instead.
Async comprehensions: [x async for x in agen()], [await f(x) for x in xs] (sequential!).
Built-ins aiter() and anext() exist since 3.10.

Async context managers

python

from contextlib import asynccontextmanager


@asynccontextmanager
async def vector_transaction(client):
    tx = await client.begin()
    try:
        yield tx
    except BaseException:
        await tx.rollback()
        raise
    else:
        await tx.commit()

AsyncExitStack manages a runtime-determined number of async resources (await stack.enter_async_context(httpx.AsyncClient())).
A class implements __aenter__/__aexit__; __aexit__ must not swallow exceptions unless intended.
Cleanup that awaits during cancellation: the finally block runs, and awaits inside it work, but a second cancellation can interrupt them; keep cleanup short or shield the critical part.

Calling blocking code

python

import asyncio
import time
from concurrent.futures import ProcessPoolExecutor


def parse_pdf(path: str) -> int:          # CPU-bound stand-in
    return sum(i * i for i in range(200_000)) % 97 + len(path)


def slow_sdk_call(x: int) -> int:         # blocking I/O stand-in
    time.sleep(0.05)
    return x * 10


async def main() -> None:
    io_results = await asyncio.gather(*(asyncio.to_thread(slow_sdk_call, i) for i in range(5)))
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        cpu_results = await asyncio.gather(*(loop.run_in_executor(pool, parse_pdf, f"doc{i}.pdf")
                                             for i in range(4)))
    print(io_results, len(cpu_results))


if __name__ == "__main__":
    asyncio.run(main())
# Output: [0, 10, 20, 30, 40] 4

to_thread (3.9) uses the loop's default ThreadPoolExecutor and copies contextvars.
The default executor has min(32, cpu_count + 4) threads; 100 concurrent to_thread calls queue behind them. Set a bigger one with loop.set_default_executor(ThreadPoolExecutor(64)) or bound with a Semaphore.
Cancellation does not stop the thread; the await is abandoned, the function runs to completion.
Process pools: functions and arguments must be picklable (module-level functions), and the if __name__ == "__main__": guard is required with spawn/forkserver start methods (forkserver is the Linux default since 3.14).
3.14: concurrent.futures.InterpreterPoolExecutor runs work in subinterpreters (separate GILs) in one process; good for pure-Python CPU work, but many C extensions do not support subinterpreters yet.

Threads calling into the loop

python

import asyncio
import threading


async def main() -> None:
    loop = asyncio.get_running_loop()
    done = asyncio.Event()
    results: list[int] = []

    async def record(x: int) -> int:
        results.append(x)
        return x * 2

    def gpu_worker() -> None:                          # e.g. a CUDA thread finishing batches
        fut = asyncio.run_coroutine_threadsafe(record(21), loop)
        print("worker got", fut.result(timeout=5))
        loop.call_soon_threadsafe(done.set)            # never call done.set() directly from a thread

    threading.Thread(target=gpu_worker, daemon=True).start()
    await done.wait()
    print("results", results)


asyncio.run(main())
# Output:
# worker got 42
# results [21]

asyncio objects are not thread-safe. From another thread only call_soon_threadsafe and run_coroutine_threadsafe are allowed.

Background tasks

python

import asyncio
import logging

logger = logging.getLogger(__name__)


class TaskSupervisor:
    """Owns fire-and-forget tasks: keeps references, logs failures, cancels on shutdown."""

    def __init__(self) -> None:
        self._tasks: set[asyncio.Task] = set()

    def spawn(self, coro, *, name: str) -> asyncio.Task:
        task = asyncio.create_task(coro, name=name)
        self._tasks.add(task)
        task.add_done_callback(self._done)
        return task

    def _done(self, task: asyncio.Task) -> None:
        self._tasks.discard(task)
        if not task.cancelled() and task.exception() is not None:
            logger.error("task %s failed", task.get_name(), exc_info=task.exception())

    async def aclose(self, timeout: float = 10.0) -> None:
        for t in self._tasks:
            t.cancel()
        if self._tasks:
            await asyncio.wait(self._tasks, timeout=timeout)

In a FastAPI app create one supervisor in lifespan and await supervisor.aclose() in its finally. Starlette's BackgroundTasks runs after the response is sent, in the same request lifecycle; it is fine for short work but not for jobs that must survive a restart (use a real queue: Celery, Arq, RQ, a DB outbox).

Eager tasks (3.12+)

python

loop = asyncio.get_running_loop()
loop.set_task_factory(asyncio.eager_task_factory)

With eager execution create_task(coro) runs the coroutine synchronously until its first real suspension. If it finishes without suspending (cache hit, validation error) the task is already done and never goes through the scheduler: big wins for memoized lookups. Costs: code before the first await now runs before create_task returns, so ordering assumptions ("the task starts after I set up X") can break. Test with it on before enabling in production. asyncio.create_eager_task_factory(custom_task_constructor) builds one for a custom Task class.

Synchronization primitives

Primitive	Use	Pitfall
`asyncio.Lock`	protect a read-modify-write that spans an await (token refresh)	not reentrant; not thread-safe
`asyncio.Semaphore(n)`	cap concurrent operations	`release()` without `acquire()` silently raises the cap; use `BoundedSemaphore`
`asyncio.Event`	one-shot or resettable signal (index ready, shutdown requested)	`clear()` races with waiters that just woke
`asyncio.Condition`	wait for a predicate under a lock	use `wait_for(predicate)`
`asyncio.Queue(maxsize)`	hand-off with backpressure	`task_done` per `get`
`asyncio.Barrier(n)` (3.11)	n tasks rendezvous (synchronized eval phases)	broken barrier raises `BrokenBarrierError`

Single-flight token refresh (one refresh even if 50 tasks see an expired token):

python

class TokenProvider:
    def __init__(self, fetch):
        self._fetch, self._token, self._expires = fetch, None, 0.0
        self._lock = asyncio.Lock()

    async def get(self) -> str:
        loop = asyncio.get_running_loop()
        if self._token and loop.time() < self._expires - 30:
            return self._token
        async with self._lock:
            if self._token and loop.time() < self._expires - 30:   # re-check after waiting
                return self._token
            self._token, ttl = await self._fetch()
            self._expires = loop.time() + ttl
            return self._token

Event loop lifecycle

python

import asyncio


async def job(n: int) -> int:
    await asyncio.sleep(0)
    return n * n


with asyncio.Runner() as runner:          # 3.11+: one loop for several top-level calls
    print(runner.run(job(3)), runner.run(job(4)))
# Output: 9 16

asyncio.run(coro, debug=None, loop_factory=None) (loop_factory 3.12+): creates a new loop, runs the coroutine, cancels remaining tasks, finalizes async generators, shuts down the default executor (3.9+) and closes the loop. Call it once per program (or per thread).

uvloop (libuv-based loop, often 2-4x faster for socket-heavy servers):

python

import asyncio

import uvloop

uvloop.run(main())                                               # uvloop >= 0.18
asyncio.run(main(), loop_factory=uvloop.new_event_loop)          # Python 3.12+

uvloop.install() sets the event loop policy; policies are deprecated in 3.14, so prefer the two forms above. Uvicorn uses uvloop automatically when installed (--loop uvloop). uvloop does not support Windows.

Jupyter / IPython: the kernel already runs a loop, so asyncio.run() raises RuntimeError: asyncio.run() cannot be called from a running event loop. Use top-level await main() in a cell (IPython autoawait). Library code that must work in both worlds should expose async functions and let the caller choose; if a sync wrapper is unavoidable, run the coroutine in a separate thread with its own loop:

python

import asyncio
import concurrent.futures


def run_sync(coro):
    """Run a coroutine from sync code, even when this thread already has a running loop (Jupyter)."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
        return ex.submit(asyncio.run, coro).result()

nest_asyncio patches the loop to allow re-entrancy; it is unmaintained and breaks with newer Python versions and uvloop, so avoid it.

Context variables

contextvars.ContextVar values (request id, tenant, trace span) are copied into each task at creation and into to_thread calls, so per-request context flows through fan-out automatically. A task's changes do not leak back to its parent. Use them instead of globals for request-scoped data in async servers.

python

import asyncio
import contextvars

request_id = contextvars.ContextVar("request_id", default="-")


async def log_step(step: str) -> str:
    await asyncio.sleep(0)
    return f"[{request_id.get()}] {step}"


async def handle(rid: str) -> list[str]:
    request_id.set(rid)
    return list(await asyncio.gather(log_step("retrieve"), log_step("generate")))


async def main() -> None:
    print(await asyncio.gather(handle("req-1"), handle("req-2")))


asyncio.run(main())
# Output: [['[req-1] retrieve', '[req-1] generate'], ['[req-2] retrieve', '[req-2] generate']]

Contents

Mental model

Coroutines, tasks and futures

TaskGroup (3.11+)

gather

TaskGroup emulation for 3.9 and 3.10

wait and as_completed

Hedged requests and racing models

Worker pools on a queue

Multi-stage pipelines

Async generators

Async context managers

Calling blocking code

Threads calling into the loop

Background tasks

Eager tasks (3.12+)

Synchronization primitives

Event loop lifecycle

Context variables

python-async-deadlock-debugger

python-beginner-concept-tutor

python-comprehension-refactorer

python-context-manager-writer

python-dataclass-converter

python-exception-handling-designer

Asyncio Patterns Catalog

#Contents

#Mental model

#Coroutines, tasks and futures

#TaskGroup (3.11+)

#gather

#TaskGroup emulation for 3.9 and 3.10

#wait and as_completed

#Hedged requests and racing models

#Worker pools on a queue

#Multi-stage pipelines

#Async generators

#Async context managers

#Calling blocking code

#Threads calling into the loop

#Background tasks

#Eager tasks (3.12+)

#Synchronization primitives

#Event loop lifecycle

#Context variables

python-async-deadlock-debugger

python-beginner-concept-tutor

python-comprehension-refactorer

python-context-manager-writer

python-dataclass-converter

python-exception-handling-designer

Contents

Mental model

Coroutines, tasks and futures

TaskGroup (3.11+)

gather

TaskGroup emulation for 3.9 and 3.10

wait and as_completed

Hedged requests and racing models

Worker pools on a queue

Multi-stage pipelines

Async generators

Async context managers

Calling blocking code

Threads calling into the loop

Background tasks

Eager tasks (3.12+)

Synchronization primitives

Event loop lifecycle

Context variables