Async Patterns with anyio: Backend-Agnostic Async in Python

Asynchronous programming in Python has evolved rapidly, but the ecosystem remains fragmented. The standard library's asyncio is the most popular, but alternatives like trio and curio offer different philosophies—structured concurrency, better cancellation, and simpler error handling. However, choosing one backend locks you into its ecosystem. Enter anyio: a backend-agnostic async library that provides a unified API across all major async backends. With anyio, you can write portable async code that runs on asyncio, trio, or curio without modification. This is a game-changer for library authors who want to support multiple backends and for developers who want to future-proof their code. In this tutorial, you'll learn the core patterns of anyio: starting an async program, structured concurrency with task groups, cancellation and timeouts, synchronization primitives, and networking. You'll also see a real-world production incident where anyio's portability saved the day. By the end, you'll be able to write robust, backend-agnostic async code that works everywhere.

Getting Started with anyio

To begin, install anyio with pip: pip install anyio. The simplest anyio program uses anyio.run() to execute an async function. This replaces asyncio.run() and works with any backend. By default, anyio uses asyncio, but you can switch to trio by passing backend='trio'. Here's a basic example:

import anyio

async def main():
    print('Hello, anyio!')
    await anyio.sleep(1)
    print('Done sleeping')

anyio.run(main)

Structured Concurrency with Task Groups

One of anyio's most powerful features is structured concurrency via TaskGroup. Unlike raw asyncio.create_task(), a task group ensures that all child tasks are completed or cancelled before the group exits. This prevents orphaned tasks and makes error handling predictable. If any task raises an exception, all sibling tasks are cancelled, and the exception is propagated. Here's an example:

import anyio

async def worker(name, delay):
    print(f'{name} starting')
    await anyio.sleep(delay)
    print(f'{name} finished')
    return f'{name} result'

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker, 'A', 2)
        tg.start_soon(worker, 'B', 1)
        tg.start_soon(worker, 'C', 3)
    print('All tasks done')

anyio.run(main)

Cancellation and Timeouts

anyio provides a unified cancellation model through CancelScope. Unlike asyncio's CancelledError, anyio's cancellation is more predictable. You can create a cancel scope and cancel it manually, or use fail_after() for timeouts. Here's an example:

import anyio

async def long_running():
    try:
        await anyio.sleep(10)
    except anyio.get_cancelled_exc_class():
        print('Cancelled!')
        raise

async def main():
    async with anyio.create_task_group() as tg:
        tg.start_soon(long_running)
        await anyio.sleep(1)
        tg.cancel_scope.cancel()
    print('Task group exited')

anyio.run(main)

Async Synchronization Primitives

anyio provides backend-agnostic synchronization primitives like Lock, Event, Semaphore, and Queue. These work identically across backends. Here's an example using anyio.Queue for a producer-consumer pattern:

import anyio

async def producer(queue, n):
    for i in range(n):
        await queue.put(i)
        print(f'Produced {i}')
        await anyio.sleep(0.5)
    await queue.put(None)  # sentinel

async def consumer(queue):
    while True:
        item = await queue.get()
        if item is None:
            break
        print(f'Consumed {item}')

async def main():
    queue = anyio.Queue(maxsize=5)
    async with anyio.create_task_group() as tg:
        tg.start_soon(producer, queue, 10)
        tg.start_soon(consumer, queue)

anyio.run(main)

Async Networking with anyio

anyio provides high-level networking APIs for TCP and UDP. The anyio.connect_tcp() and anyio.create_tcp_listener() functions are backend-agnostic. Here's a simple echo server and client:

import anyio

async def main():
    stream = await anyio.connect_tcp('127.0.0.1', 12345)
    async with stream:
        await stream.send(b'Hello')
        response = await stream.receive(1024)
        print(f'Received: {response}')

anyio.run(main)

Running Blocking Code with anyio

anyio provides to_thread.run_sync() to run blocking code in a separate thread without blocking the event loop. This is essential for integrating with synchronous libraries. Here's an example:

import anyio
import time

def blocking_io():
    time.sleep(2)
    return 'result'

async def main():
    result = await anyio.to_thread.run_sync(blocking_io)
    print(f'Got {result}')

anyio.run(main)