Advanced asyncio Patterns for Production Python

Most asyncio tutorials stop at await and gather. Production code needs more: controlled concurrency, graceful cancellation, error handling that does not swallow exceptions, and patterns that prevent resource leaks. This article covers the asyncio patterns that matter when your code runs in production.

TaskGroup: Structured Concurrency

Python 3.11 introduced TaskGroup, which ensures that all spawned tasks complete (or are cancelled) before execution continues. Unlike gather, a TaskGroup cancels remaining tasks if any task raises an exception.

import asyncio

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # Simulate network call
    return {"id": user_id, "name": f"User {user_id}"}

async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    results = []

    async with asyncio.TaskGroup() as tg:
        for uid in user_ids:
            task = tg.create_task(fetch_user(uid))
            results.append(task)

    return [task.result() for task in results]

async def main():
    users = await fetch_all_users([1, 2, 3, 4, 5])
    for user in users:
        print(user)

asyncio.run(main())

TaskGroup Error Handling

If any task fails, the group cancels all other tasks and raises an ExceptionGroup:

import asyncio

async def risky_task(n: int) -> int:
    await asyncio.sleep(0.1)
    if n == 3:
        raise ValueError(f"Task {n} failed")
    return n * 10

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = [tg.create_task(risky_task(i)) for i in range(5)]
    except* ValueError as eg:
        for exc in eg.exceptions:
            print(f"Caught: {exc}")
    # All other tasks were cancelled when task 3 failed

asyncio.run(main())

The except* syntax (Python 3.11+) handles exception groups. Each exception in the group can be processed individually.

Semaphores: Limiting Concurrency

Launching thousands of concurrent tasks can overwhelm servers or hit rate limits. A semaphore limits how many tasks run simultaneously:

import asyncio
import aiohttp

async def fetch_url(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> str:
    async with sem:  # At most N concurrent requests
        async with session.get(url) as response:
            return await response.text()

async def fetch_many(urls: list[str], max_concurrent: int = 10) -> list[str]:
    sem = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url, sem) for url in urls]
        return await asyncio.gather(*tasks)

async def main():
    urls = [f"https://httpbin.org/get?id={i}" for i in range(100)]
    results = await fetch_many(urls, max_concurrent=20)
    print(f"Fetched {len(results)} pages")

asyncio.run(main())

Even though 100 tasks are created, only 20 execute at any given time.

Timeouts

Use asyncio.timeout (Python 3.11+) or asyncio.wait_for to prevent tasks from running forever:

import asyncio

async def slow_operation() -> str:
    await asyncio.sleep(10)
    return "done"

async def main():
    # Python 3.11+ context manager approach
    try:
        async with asyncio.timeout(2.0):
            result = await slow_operation()
    except TimeoutError:
        print("Operation timed out after 2 seconds")

    # Alternative: wait_for (works in older Python versions)
    try:
        result = await asyncio.wait_for(slow_operation(), timeout=2.0)
    except asyncio.TimeoutError:
        print("Operation timed out after 2 seconds")

asyncio.run(main())

Per-Task Timeouts in a Group

import asyncio

async def fetch_with_timeout(url: str, timeout: float) -> str | None:
    try:
        async with asyncio.timeout(timeout):
            await asyncio.sleep(1)  # Simulate fetch
            return f"Result from {url}"
    except TimeoutError:
        return None  # Graceful degradation

async def main():
    urls = ["fast.com", "slow.com", "medium.com"]
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_with_timeout(url, 0.5)) for url in urls]

    results = [t.result() for t in tasks]
    successful = [r for r in results if r is not None]
    print(f"{len(successful)} of {len(urls)} succeeded")

asyncio.run(main())

Retry Logic

Network operations fail. A retry wrapper with exponential backoff handles transient errors:

import asyncio
import random

async def retry(
    coro_func,
    *args,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
    retryable_exceptions: tuple = (Exception,),
):
    for attempt in range(max_retries + 1):
        try:
            return await coro_func(*args)
        except retryable_exceptions as e:
            if attempt == max_retries:
                raise
            delay = min(base_delay * (2 ** attempt), max_delay)
            jitter = random.uniform(0, delay * 0.1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay + jitter:.1f}s")
            await asyncio.sleep(delay + jitter)

# Usage
async def flaky_api_call(endpoint: str) -> dict:
    # Simulated flaky call
    if random.random() < 0.5:
        raise ConnectionError("Server unavailable")
    return {"status": "ok"}

async def main():
    result = await retry(
        flaky_api_call,
        "/api/data",
        max_retries=3,
        retryable_exceptions=(ConnectionError, TimeoutError),
    )
    print(result)

asyncio.run(main())

Async Generators

Async generators let you yield values from asynchronous operations, which is useful for streaming data:

import asyncio
from typing import AsyncIterator

async def paginate(base_url: str, page_size: int = 100) -> AsyncIterator[dict]:
    page = 1
    while True:
        # Simulate fetching a page
        await asyncio.sleep(0.1)
        items = [{"id": (page - 1) * page_size + i} for i in range(page_size)]

        if not items:
            break

        for item in items:
            yield item

        if len(items) < page_size:
            break
        page += 1

async def main():
    count = 0
    async for item in paginate("https://api.example.com/items", page_size=50):
        count += 1
        if count >= 150:
            break
    print(f"Processed {count} items")

asyncio.run(main())

Async Context Managers

Wrap async setup and teardown in context managers to prevent resource leaks:

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_connection(host: str, port: int):
    reader, writer = await asyncio.open_connection(host, port)
    try:
        yield reader, writer
    finally:
        writer.close()
        await writer.wait_closed()

async def main():
    async with managed_connection("example.com", 80) as (reader, writer):
        writer.write(b"GET / HTTP/1.0\r\nHost: example.com\r\n\r\n")
        await writer.drain()
        response = await reader.read(1024)
        print(response[:100])

asyncio.run(main())

Graceful Shutdown

Production services need to clean up when receiving shutdown signals:

import asyncio
import signal

async def worker(name: str):
    try:
        while True:
            print(f"{name}: working...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print(f"{name}: cleaning up...")
        await asyncio.sleep(0.1)  # Cleanup work
        print(f"{name}: done")

async def main():
    loop = asyncio.get_running_loop()
    shutdown_event = asyncio.Event()

    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, shutdown_event.set)

    tasks = [
        asyncio.create_task(worker("worker-1")),
        asyncio.create_task(worker("worker-2")),
    ]

    await shutdown_event.wait()
    print("Shutting down...")

    for task in tasks:
        task.cancel()

    await asyncio.gather(*tasks, return_exceptions=True)
    print("Shutdown complete")

asyncio.run(main())

Key Takeaways

Production asyncio code needs more than gather. Use TaskGroup for structured concurrency that cleans up on failure. Use semaphores to control concurrent access to limited resources. Add timeouts to every external call. Implement retry logic with exponential backoff. Use async context managers to prevent resource leaks, and handle shutdown signals gracefully. These patterns are the difference between a demo and a production service.