Python Async IO: Master Concurrency Patterns

Programming
Date:September 11, 2026
Topic:
Python Async IO: Master Concurrency Patterns
2 min read

Python async IO has grown up. The gather-and-pray era is over. Python 3.11+ introduced TaskGroup and asyncio.timeout, and 3.14 adds eager task starts, queue shutdown APIs, and finer runner control. If you're still writing try/except around gather(), you're working too hard and hiding bugs.

Structured concurrency is the new default

TaskGroup replaces gather() for nearly every use case. It guarantees that if one child fails, the others are cancelled automatically. No more orphaned tasks leaking memory or holding connections open.

python
async def fetch_all(urls: list[str]) -> list[dict]:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]
    return [t.result() for t in tasks]
💡
TipTaskGroup.__aexit__ waits for all tasks. If any raises, it cancels the rest and re-raises the first exception. Use except* (PEP 654) to handle multiple failures.

Timeouts that actually work

asyncio.timeout() (3.11+) and timeout_at() are context managers, not decorator hacks. They cancel the enclosed block cleanly on expiry.

python
async def with_deadline():
    async with asyncio.timeout(5.0):
        await slow_operation()
    # cancelled automatically if over 5s

Eager task starts in 3.14

Python 3.14 adds TaskGroup.create_task(eager_start=True). The coroutine runs synchronously until its first await, reducing scheduler overhead for tasks that often complete immediately (cache hits, fast paths).

python
async with asyncio.TaskGroup() as tg:
    tg.create_task(fetch_cached(key), eager_start=True)

Queues with shutdown semantics

asyncio.Queue now supports shutdown() and shutted_down(). Producers call shutdown(); consumers detect it via queue.get() raising QueueShutDown. No more sentinel values or custom protocols.

python
async def producer(q: asyncio.Queue):
    for item in items:
        await q.put(item)
    q.shutdown()

async def consumer(q: asyncio.Queue):
    while True:
        try:
            item = await q.get()
        except asyncio.QueueShutDown:
            break
        await process(item)

Runner control for apps and tests

asyncio.Runner (3.11+) lets you reuse a loop with a custom context. In 3.14, runner.run() accepts a timeout and returns the loop for inspection. Great for test fixtures and embedding.

python
runner = asyncio.Runner(loop_factory=uvloop.new_event_loop)
try:
    await runner.run(main(), timeout=30.0)
finally:
    runner.close()

Common pitfalls still worth avoiding

Anti-patternFix
await gather(*tasks) without try/exceptUse TaskGroup
Creating tasks but not awaiting themTrack in TaskGroup or list
Blocking calls in async functionsRun in executor: await loop.run_in_executor(None, blocking_fn)
Ignoring CancelledErrorLet it propagate; cleanup in finally

Migration checklist

1. Replace gather() with TaskGroup. 2. Wrap network boundaries with asyncio.timeout(). 3. Use eager_start for cache-heavy workloads. 4. Replace sentinel queues with queue.shutdown(). 5. Adopt Runner for integration tests. 6. Run your test suite with PYTHONASYNCIODEBUG=1 to catch unawaited coroutines.

"

Structured concurrency isn't syntax sugar. It's the difference between a server that recovers from overload and one that leaks until OOM.

Guido van Rossum (paraphrased)


ℹ️
NoteStart small: pick one service endpoint, wrap its outbound calls in TaskGroup and timeout. Measure error rates and tail latency. Then expand.
Share𝕏 Twitterin LinkedInin Whatsapp