Python asyncio Patterns [2026] Async Engineering Reference

As of April 6, 2026, the stable Python docs track Python 3.14.3, and that matters for async teams: TaskGroup is mature, create_task() now exposes eager_start, and Queue.shutdown() gives producer-consumer pipelines a cleaner exit path.

This reference is optimized for fast lookup, not narrative teaching. Use the search box to filter APIs, patterns, and operational notes. If you want to clean up examples before sharing them internally, the Code Formatter is a good companion for polishing Python snippets.

What Changed in 2026

• Python 3.14 extends asyncio.create_task() and TaskGroup.create_task() so keyword arguments pass through to loop task creation, including eager_start.
• Queue.shutdown(), added in Python 3.13, is now a practical default for graceful worker shutdown instead of sentinel-only designs.
• TaskGroup remains the preferred structured-concurrency primitive for related work that should fail together.
• asyncio.Runner is still the right tool when you need repeated top-level async calls under one managed loop.

Quick Start

If you need one default template for 2026, start here.

import asyncio

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

async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        a = tg.create_task(fetch("a", 0.2))
        b = tg.create_task(fetch("b", 0.4))

    print(a.result(), b.result())

if __name__ == "__main__":
    asyncio.run(main())

Why this shape works: asyncio.run() owns loop lifecycle, TaskGroup gives structured cancellation, and task handles are preserved for result access and observability.

Commands by Purpose

Use live filter: press / to focus search, type any API or pattern name, then press Esc to clear.

Keyboard Shortcuts

These shortcuts are wired into the page script below for quick reference scanning.

Configuration

These are the knobs that matter most in real services.

Runtime Debugging

• Use asyncio.run(main(), debug=True) for one-shot debugging.
• Set PYTHONASYNCIODEBUG=1 in environments where you want debug mode enabled broadly.
• Name important tasks with name= for logs and introspection.

import asyncio

async def worker() -> None:
    await asyncio.sleep(0.1)

async def main() -> None:
    task = asyncio.create_task(worker(), name="cache-refresh")
    await task

asyncio.run(main(), debug=True)

Runner for Embedded or Repeated Calls

asyncio.Runner is cleaner than manually creating and closing loops when a synchronous host needs to invoke async code repeatedly.

import asyncio

async def ping(value: int) -> int:
    await asyncio.sleep(0)
    return value + 1

with asyncio.Runner(debug=False) as runner:
    print(runner.run(ping(1)))
    print(runner.run(ping(2)))

Eager Task Start

eager_start can reduce scheduling overhead when coroutines often complete synchronously, but it is a semantic change, not a free speed flag. Execution order can shift.

• Good fit: cache-heavy coroutines, memoized lookups, cheap short-circuit paths.
• Bad fit: code that implicitly depends on delayed scheduling order.

Core Patterns

Structured Fan-Out with TaskGroup

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.05)
    return {"user_id": user_id}

async def fetch_permissions(user_id: int) -> list[str]:
    await asyncio.sleep(0.05)
    return ["read", "write"]

async def load_profile(user_id: int) -> tuple[dict, list[str]]:
    async with asyncio.TaskGroup() as tg:
        user_task = tg.create_task(fetch_user(user_id))
        perms_task = tg.create_task(fetch_permissions(user_id))

    return user_task.result(), perms_task.result()

Prefer this over manual task lists when tasks are siblings in one logical operation. If one fails, the group cancels the rest.

Scoped Deadlines with asyncio.timeout()

import asyncio

async def call_backend() -> str:
    await asyncio.sleep(0.2)
    return "ok"

async def main() -> None:
    try:
        async with asyncio.timeout(0.1):
            print(await call_backend())
    except TimeoutError:
        print("deadline exceeded")

asyncio.run(main())

Use asyncio.timeout() when the deadline applies to a whole block. Use wait_for() when the deadline belongs to one awaitable.

Bounded Concurrency with Semaphore

import asyncio

sem = asyncio.Semaphore(10)

async def fetch(url: str) -> str:
    async with sem:
        await asyncio.sleep(0.05)
        return url

async def main(urls: list[str]) -> list[str]:
    return await asyncio.gather(*(fetch(url) for url in urls))

This is the standard pattern for rate-limited APIs, connection pools, and any async workload that can overwhelm a dependency.

Worker Pools with Queue

import asyncio

async def worker(name: str, queue: asyncio.Queue[int]) -> None:
    try:
        while True:
            item = await queue.get()
            try:
                await asyncio.sleep(0.05)
                print(name, item)
            finally:
                queue.task_done()
    except asyncio.QueueShutDown:
        return

async def main() -> None:
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=100)

    async with asyncio.TaskGroup() as tg:
        for idx in range(3):
            tg.create_task(worker(f"w{idx}", queue))

        for item in range(10):
            await queue.put(item)

        await queue.join()
        queue.shutdown()

In 2026, Queue.shutdown() is the cleanest close signal for many pipelines. Sentinel values still work, but the built-in shutdown path is more explicit.

Advanced Usage

Bridge Blocking Code with to_thread()

If the code blocks on file I/O, compression, SDK calls, or legacy libraries, move it out of the event loop.

import asyncio
import time

def blocking_lookup() -> str:
    time.sleep(0.2)
    return "done"

async def main() -> None:
    result = await asyncio.to_thread(blocking_lookup)
    print(result)

Stream Results Early with as_completed()

Use gather() when you want ordered aggregation. Use as_completed() when you want the fastest completed result first, such as speculative fan-out or partial UI hydration.

Protect Critical Cleanup with shield()

asyncio.shield() is useful when outer request cancellation should not interrupt a narrow commit or flush sequence. Use it sparingly. Shielding too much defeats cooperative cancellation.

Backpressure on Writes with drain()

For stream writers, always treat writer.drain() as part of the write path. Skipping it under load is how fast producers outrun socket buffers.

Cross-Thread Submission

asyncio.run_coroutine_threadsafe() is the escape hatch for systems where a non-async thread must inject work into a running loop. Keep that boundary small and explicit.

Pitfalls

• Do not call asyncio.run() from inside an already-running event loop.
• Do not swallow CancelledError unless you fully understand the cancellation state you are suppressing.
• Do not lose task references. Unreferenced tasks may disappear before completion because the loop keeps weak references.
• Do not use gather() everywhere by habit. If the tasks are structurally related, use TaskGroup.
• Do not run blocking libraries directly in async code. Use to_thread() or an executor.
• Do not forget queue.task_done() after get() in worker pipelines.
• Do not assume eager_start is behavior-neutral. It can change task ordering.

Resources

• Official asyncio library reference
• Official coroutines and tasks reference
• Official asyncio queue reference
• Official asyncio streams reference

One last operational note: when async examples include production payloads, scrub secrets before sharing traces or screenshots. The Data Masking Tool is useful for that workflow.