Python AnyIO Guide: Mastering Structured Concurrency Over Raw asyncio

Comparing the 3 Async Paradigms in Python

Writing async code in Python starts out simple with basic async/await syntax. But as systems scale—such as a worker handling 500 requests per second—the fundamental differences between concurrency models quickly become apparent:

• 1. Traditional asyncio (Unstructured Concurrency): Spawns tasks via asyncio.create_task() or asyncio.gather(). Tasks run in a “fire-and-forget” fashion, executing independently and detached from the caller’s lifecycle.
• 2. Trio (Pure Structured Concurrency): Introduces structured concurrency principles to Python. Every child task must be spawned and completed within an explicit scope (a Nursery), completely eliminating orphan tasks.
• 3. AnyIO (Cross-Platform Structured Concurrency): Brings Trio’s structured concurrency standards to the asyncio ecosystem. You gain safe task groups while retaining access to the vast asyncio library ecosystem without being locked into a single runtime.

Pros and Cons from a Production Perspective

An automation pipeline I once maintained grew from 200 lines to over 3,000 lines. That was when subtle race conditions and memory leaks began creeping in.

1. Raw asyncio: Convenient but Full of Pitfalls

asyncio comes built into the standard library with extensive documentation. However, its loose lifecycle management carries significant risks:

• Tasks collected by Garbage Collector: Calling create_task() without retaining a reference allows the garbage collector (GC) to clean up the task mid-execution. The official Python docs added an explicit warning about this classic bug in Python 3.10.
• Resource-hogging orphan tasks: In a group of 10 tasks, if task 2 fails, the remaining 9 continue running in the background. Dozens of database connections or open sockets can linger indefinitely.
• Swallowing cancellation errors: A broad except Exception: block can inadvertently catch asyncio.CancelledError, causing application shutdown routines to hang completely.

2. Trio: An Architectural Gold Standard, but Ecosystem-Constrained

Trio’s nursery pattern provides rigorous concurrency boundaries and elegant exception handling. Its primary drawback is ecosystem compatibility. Embedding Trio into services built around Uvicorn, Celery, or libraries tightly coupled with the asyncio event loop is virtually impossible.

3. AnyIO: A Pragmatic Abstraction Layer

AnyIO functions as an abstraction layer running seamlessly across both asyncio and trio backends. Prominent frameworks such as FastAPI (via Starlette) and HTTPX rely on AnyIO under the hood for their asynchronous operations.

• Pros: Delivers strict TaskGroups, flexible cancel scopes, stream-based data synchronization, and 100% compatibility with the default event loop.
• Cons: Requires an external dependency (pip install anyio). Developers must also break the habit of spawning arbitrary tasks and adopt scope-based thinking.

When Should You Choose Which?

• Stick with raw asyncio: Best suited for simple scraping scripts or fetching 20–30 static web pages using a handful of independent coroutines.
• Adopt AnyIO immediately: Essential for background workers processing thousands of jobs, high-concurrency microservices, or any application requiring guaranteed resource cleanup upon client disconnections.

Practical Guide to Implementing AnyIO

Step 1: Install the Library

pip install anyio

Step 2: Run with anyio.run

Instead of asyncio.run(), bootstrap your application using anyio.run(). AnyIO defaults to the asyncio backend:

import anyio

async def main():
    print("System starting...")
    await anyio.sleep(1)
    print("Task completed!")

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

Step 3: Safely Group Tasks with TaskGroup

A core rule in AnyIO is to always group related tasks inside create_task_group(). The async with block guarantees that execution exits only when all child tasks have finished. If one task crashes, all remaining tasks in the group are immediately cancelled:

import anyio

async def fetch_service_data(service_id: int):
    print(f"[Service {service_id}] Fetching data...")
    await anyio.sleep(service_id)
    if service_id == 2:
        raise ValueError("Service 2 connection lost unexpectedly!")
    print(f"[Service {service_id}] Completed.")

async def main():
    try:
        async with anyio.create_task_group() as tg:
            for i in [1, 2, 3]:
                tg.start_soon(fetch_service_data, i)
    except* ValueError as eg:
        # On Python 3.11+, AnyIO uses native ExceptionGroup
        print(f"Caught grouped exceptions from TaskGroup: {eg.exceptions}")

anyio.run(main)

The benefit is clear: when Service 2 raises a ValueError, AnyIO immediately cancels the still-running Service 3. You no longer have to worry about Task 3 running aimlessly out of control.

Step 4: Clean Timeout Control with move_on_after

Instead of wrapping asyncio.wait_for() in cumbersome try/except blocks, AnyIO offers the much more convenient move_on_after() context manager:

import anyio

async def long_running_task():
    # Limit this entire block to a maximum of 2.5 seconds
    with anyio.move_on_after(2.5) as scope:
        print("Starting heavy data sync...")
        await anyio.sleep(5)
        print("This line will never be printed if timed out.")

    if scope.cancelled_caught:
        print("Time limit exceeded! Safely stopped the task.")

anyio.run(long_running_task)

Unlike fail_after() which raises a TimeoutError, move_on_after() silently exits the block when the time limit expires. You simply check the scope.cancelled_caught flag to handle subsequent logic.

Step 5: Safely Pass Data with Memory Object Streams

For Producer-Consumer patterns, prefer AnyIO streams over asyncio.Queue. Their key advantage is explicit, two-way stream closure handling:

import anyio

async def producer(send_stream):
    async with send_stream:
        for item in ["log_1.json", "log_2.json", "log_3.json"]:
            print(f"Enqueued: {item}")
            await send_stream.send(item)
            await anyio.sleep(0.5)

async def consumer(receive_stream):
    async with receive_stream:
        async for message in receive_stream:
            print(f"Processed successfully: {message}")

async def main():
    send_stream, receive_stream = anyio.create_memory_object_stream(max_buffer_size=10)
    async with anyio.create_task_group() as tg:
        tg.start_soon(producer, send_stream)
        tg.start_soon(consumer, receive_stream)

anyio.run(main)

Production Tips When Working with AnyIO

• Avoid detached tasks: Pass the task_group directly into your service layer whenever you need to spawn sub-tasks. This keeps the task tree transparent and structured.
• Offload blocking calls to worker threads: When reading a 500MB file or invoking synchronous libraries, use await anyio.to_thread.run_sync(blocking_func, arg) instead of manually managing a ThreadPoolExecutor.
• Handle errors cleanly with ExceptionGroup: On Python 3.11+, use except* syntax to unpack and handle individual exceptions raised by concurrent tasks.