1) Mental Models

• asyncio: Batteries-in-stdlib, widespread lib support, explicit task management.
• Trio: Structured concurrency first; nurseries force predictable lifecycles.
• AnyIO: Unified API that targets asyncio/Trio under the hood; great for libraries.

2) Structured Concurrency with asyncio (3.11+)

import asyncio

async def worker(i):
    await asyncio.sleep(0.1)
    return i * 2

async def main():
    results = []
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(worker(i)) for i in range(10)]
    # TaskGroup waits; exceptions propagate
    # gather results after completion (store via closures or queues)
    print("Done")

asyncio.run(main())
      

Use when you need stdlib, broad compatibility, and TaskGroup semantics.

3) Trio’s Nurseries (ergonomic cancellation)

import trio

async def worker(i):
    await trio.sleep(0.1)
    return i * 2

async def main():
    async with trio.open_nursery() as nursery:
        for i in range(10):
            nursery.start_soon(worker, i)  # heavy use of cancellation propagation

trio.run(main)
      

Use when you want first-class structured concurrency and superb cancellation ergonomics.

4) AnyIO (write-once, run-anywhere)

import anyio

async def fetch(i):
    await anyio.sleep(0.1)
    return i

async def main():
    async with anyio.create_task_group() as tg:
        for i in range(5):
            tg.start_soon(fetch, i)

anyio.run(main)
      

Use when you’re authoring libraries or want runtime flexibility.

5) Timeouts & Cancellation Patterns

# asyncio timeout
import asyncio
async def bounded():
    try:
        await asyncio.wait_for(asyncio.sleep(2), timeout=0.5)
    except asyncio.TimeoutError:
        ...

# Trio timeout
import trio
async def bounded_trio():
    with trio.move_on_after(0.5) as cancel_scope:
        await trio.sleep(2)
        if cancel_scope.cancelled_caught:
            ...
      

6) Graceful Shutdown

# asyncio: cancel all tasks and drain
import asyncio, signal

async def main():
    loop = asyncio.get_running_loop()
    stop = asyncio.Event()
    loop.add_signal_handler(signal.SIGTERM, stop.set)
    try:
        await stop.wait()
    finally:
        # flush queues, close clients, etc.
        ...

asyncio.run(main())
      

7) When to Choose Which

• Web frameworks / ecosystem breadth: asyncio
• Greenfield services / correctness-first: Trio
• Libraries / portability: AnyIO