Python Async Patterns: Beyond the Basics

Python's async ecosystem has matured significantly since asyncio entered the standard library. Yet many developers stop at basic async/await syntax without exploring the patterns that make asynchronous Python genuinely useful. This guide covers practical patterns for building concurrent Python applications that go beyond introductory tutorials.

The Event Loop Foundation

Every asyncio program runs on an event loop that schedules and executes coroutines. Understanding the event loop's behavior explains why certain patterns work and others deadlock.

import asyncio

async def main():
    print("Starting")
    await asyncio.sleep(1)  # Yields control back to the event loop
    print("Done")

asyncio.run(main())  # Creates event loop, runs coroutine, closes loop

The critical insight: await doesn't mean "wait here." It means "yield control to the event loop, which may run other coroutines, and resume this coroutine when the awaited result is ready." Blocking calls like time.sleep() or synchronous I/O freeze the entire event loop because they never yield control.

# Wrong: blocks the event loop for 5 seconds
async def bad_example():
    import time
    time.sleep(5)  # No other coroutines can run during this

# Right: run blocking code in a thread pool
async def good_example():
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, time.sleep, 5)

asyncio.gather vs TaskGroup

asyncio.gather runs multiple coroutines concurrently and collects their results. It has been the standard approach since Python 3.4.

async def fetch_user(user_id: int) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(f"https://api.example.com/users/{user_id}") as resp:
            return await resp.json()

async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    return await asyncio.gather(*(fetch_user(uid) for uid in user_ids))

Python 3.11 introduced TaskGroup, which provides structured concurrency and better error handling:

async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    results = []
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch_user(uid)) for uid in user_ids]
    return [task.result() for task in tasks]

Key differences:

• Error handling: gather with return_exceptions=False cancels remaining tasks on the first exception but can leave orphaned tasks. TaskGroup guarantees all tasks are cancelled and awaited before propagating exceptions via ExceptionGroup.
• Cancellation: TaskGroup properly cancels all tasks when any task fails. gather requires explicit cleanup.
• Nesting: TaskGroup contexts can nest naturally. Nested gather calls require careful error propagation.

Prefer TaskGroup for new code targeting Python 3.11+. Use gather when you need return_exceptions=True behavior or support older Python versions.

Async Queues for Producer-Consumer Patterns

asyncio.Queue enables clean producer-consumer architectures where tasks are decoupled from their processing.

import asyncio
from dataclasses import dataclass

@dataclass
class Job:
    url: str
    retries: int = 0

async def producer(queue: asyncio.Queue, urls: list[str]):
    for url in urls:
        await queue.put(Job(url=url))
    # Signal workers to stop
    for _ in range(NUM_WORKERS):
        await queue.put(None)

async def worker(queue: asyncio.Queue, results: list):
    while True:
        job = await queue.get()
        if job is None:
            break
        try:
            result = await process_url(job.url)
            results.append(result)
        except Exception as e:
            if job.retries < 3:
                job.retries += 1
                await queue.put(job)  # Re-queue for retry
        finally:
            queue.task_done()

NUM_WORKERS = 10

async def main(urls: list[str]):
    queue = asyncio.Queue(maxsize=100)  # Backpressure at 100 pending items
    results = []

    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(queue, urls))
        for _ in range(NUM_WORKERS):
            tg.create_task(worker(queue, results))

return results

Setting `maxsize` on the queue provides backpressure: the producer blocks when the queue is full, preventing memory exhaustion when production outpaces consumption.

## Error Handling in Async Code

Async error handling has unique challenges. Exceptions in fire-and-forget tasks can be silently swallowed if not properly handled.

```python
# Dangerous: exception in background_task is never observed
async def risky():
    task = asyncio.create_task(background_task())
    # If background_task raises, the exception is lost
    # Python will emit "Task exception was never retrieved" warning

# Safe: always await or attach callbacks to tasks
async def safe():
    task = asyncio.create_task(background_task())
    task.add_done_callback(handle_task_result)

def handle_task_result(task: asyncio.Task):
    if task.cancelled():
        return
    if exc := task.exception():
        logger.error(f"Background task failed: {exc}")

For ExceptionGroup handling with TaskGroup (Python 3.11+):

async def resilient_fetch(urls: list[str]) -> list[dict]:
    results = {}
    try:
        async with asyncio.TaskGroup() as tg:
            tasks = {url: tg.create_task(fetch(url)) for url in urls}
    except* aiohttp.ClientError as eg:
        for exc in eg.exceptions:
            logger.warning(f"HTTP error: {exc}")
    except* asyncio.TimeoutError as eg:
        logger.warning(f"{len(eg.exceptions)} requests timed out")

    return {url: task.result() for url, task in tasks.items() if not task.cancelled() and task.exception() is None}

The except* syntax lets you catch specific exception types from an ExceptionGroup while letting others propagate.

Structured Concurrency with Timeouts

Structured concurrency means that all spawned tasks have a clearly defined lifetime. Combine TaskGroup with asyncio.timeout for robust concurrent operations:

async def fetch_with_fallback(primary_url: str, fallback_url: str) -> dict:
    try:
        async with asyncio.timeout(5):
            return await fetch(primary_url)
    except TimeoutError:
        logger.warning(f"Primary URL timed out, trying fallback")
        return await fetch(fallback_url)

async def parallel_fetch_with_deadline(urls: list[str], deadline: float) -> list[dict]:
    results = []
    try:
        async with asyncio.timeout(deadline):
            async with asyncio.TaskGroup() as tg:
                tasks = [tg.create_task(fetch(url)) for url in urls]
            results = [t.result() for t in tasks]
    except TimeoutError:
        logger.warning("Deadline exceeded, returning partial results")
        results = [t.result() for t in tasks if t.done() and not t.cancelled()]
    return results

Practical aiohttp Patterns

aiohttp is the most widely used async HTTP client/server library. A production-ready client session handles connection pooling, timeouts, and retries:

import aiohttp
from aiohttp import ClientTimeout

async def create_api_client() -> aiohttp.ClientSession:
    timeout = ClientTimeout(total=30, connect=5, sock_read=10)
    connector = aiohttp.TCPConnector(
        limit=100,          # Max concurrent connections
        limit_per_host=10,  # Max per-host connections
        ttl_dns_cache=300,  # DNS cache TTL in seconds
    )
    return aiohttp.ClientSession(
        timeout=timeout,
        connector=connector,
        headers={"User-Agent": "MyApp/1.0"},
    )

async def fetch_with_retry(session: aiohttp.ClientSession, url: str, max_retries: int = 3) -> dict:
    for attempt in range(max_retries):
        try:
            async with session.get(url) as response:
                response.raise_for_status()
                return await response.json()
        except aiohttp.ClientResponseError as e:
            if e.status < 500 or attempt == max_retries - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

Always reuse ClientSession instances rather than creating one per request. The session maintains a connection pool, and creating sessions repeatedly wastes resources and leaks connections.

Semaphores for Rate Limiting

When calling external APIs with rate limits, use asyncio.Semaphore to cap concurrent requests:

async def rate_limited_fetch(urls: list[str], max_concurrent: int = 5) -> list[dict]:
    semaphore = asyncio.Semaphore(max_concurrent)

    async def bounded_fetch(url: str) -> dict:
        async with semaphore:
            return await fetch(url)

    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(bounded_fetch(url)) for url in urls]
    return [t.result() for t in tasks]

This ensures no more than max_concurrent requests run simultaneously, regardless of how many URLs are queued. For token-bucket rate limiting with bursting capability, consider the aiolimiter library.

Async Python excels at I/O-bound workloads: HTTP clients, database queries, file operations, and message queue consumers. For CPU-bound work, use loop.run_in_executor with a ProcessPoolExecutor to offload computation to separate processes without blocking the event loop.