Top 7 Python Concurrency Plays

Practical Trio, AnyIO, and async-generator patterns that make your services faster, safer, and easier to reason about.

Seven proven Python concurrency plays using Trio, AnyIO, and async generators — structured tasks, timeouts, backpressure, and graceful shutdowns that ship.

You don’t need “more threads.”
You need fewer surprises.

Get Thinking Loop’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Modern Python concurrency is less about clever tricks and more about structured, predictable flows. Trio and AnyIO give you that spine; async generators give you streaming without hurting readability. Below are seven plays I reach for in real services — each one battle-tested, bite-sized, and boring in the best way.

1) Structured Concurrency with Task Groups (Trio & AnyIO)

Why it matters: Spawning work is easy; collecting it — and its failures — is hard. Structured concurrency ensures tasks start together, fail together, and shut down together.

Trio

import trio
import httpx

async def fetch(client, url):
    r = await client.get(url, timeout=5)
    return url, r.status_code

async def main(urls):
    async with httpx.AsyncClient() as client:
        async with trio.open_nursery() as nursery:
            results = {}
            for u in urls:
                nursery.start_soon(
                    lambda uu=u: results.update({uu: trio.move_on_after(0) or (await fetch(client, uu))[1]})
                )
        return results

trio.run(main, ["https://example.com", "https://httpbin.org/get"])

AnyIO (portable across Trio/asyncio)

import anyio, httpx

async def main(urls):
    async with httpx.AsyncClient() as client:
        async with anyio.create_task_group() as tg:
            results = {}
            async def worker(u):
                r = await client.get(u, timeout=5)
                results[u] = r.status_code
            for u in urls:
                tg.start_soon(worker, u)
        return results

anyio.run(main, ["https://example.com", "https://httpbin.org/get"])

Takeaway: Task groups make concurrency composable. No orphan tasks, no hidden exceptions, and cancellation is a first-class citizen.

2) Deadlines and Cancellation That Propagate

Why it matters: Timeouts on a single call are not enough. You want a deadline that flows through the call graph so sub-tasks don’t run past the point of usefulness.

Trio deadline

import trio

async def call_with_deadline(deadline):
    with trio.move_on_after(deadline) as cancel_scope:
        await trio.sleep(10)  # expensive
        return "done"
    # cancelled if we exit the 'with' early
    return None

AnyIO timeout

import anyio

async def call_with_timeout():
    try:
        with anyio.fail_after(2):
            await anyio.sleep(10)
    except TimeoutError:
        ...

Practical tip: Prefer deadline semantics (time remaining) over per-call timeouts. Logging “deadline in 153 ms” helps you choose fallbacks without guesswork.

3) Capacity Limiters: Concurrency Without the Stampede

Why it matters: Unbounded concurrency crushes databases, saturates sockets, and triggers rate limits. Capacity limiters encode safety at the edges.

import anyio, httpx
from anyio import CapacityLimiter

limiter = CapacityLimiter(20)  # at most 20 concurrent calls

async def guarded_fetch(client, url):
    async with limiter:
        return await client.get(url)

async def main(urls):
    async with httpx.AsyncClient() as client:
        async with anyio.create_task_group() as tg:
            for u in urls:
                tg.start_soon(guarded_fetch, client, u)

Where to place it: At boundaries — HTTP clients, DB pools, S3 uploads. If a system downstream has a QPS budget, encode it here.

4) Async Generators for Backpressure & Streaming

Why it matters: “Just gather everything” looks fine until memory spikes. Async generators give you bounded memory and natural backpressure.

Stream, transform, and yield

import anyio, aiofiles

async def read_lines(path):
    async with aiofiles.open(path, "r") as f:
        async for line in f:
            yield line.strip()

async def batched(iterable, n=100):
    batch = []
    async for item in iterable:
        batch.append(item)
        if len(batch) >= n:
            yield batch
            batch = []
    if batch:
        yield batch

async def pipeline(path):
    async for group in batched(read_lines(path), n=500):
        # process 500 at a time; downstream pull controls the pace
        yield [x.upper() for x in group]

Rule of thumb: If you can write it as async for ..., you probably should. It makes slow producers and slow consumers play nicely.

5) Graceful Shutdown with Cancel Scopes and Aclose

Why it matters: Termination is where bugs hide. Ensure generators close and tasks unwind so you don’t leak sockets, files, or cloud handles.

import trio

async def producer(send_channel):
    async with send_channel:
        for i in range(1000):
            await send_channel.send(i)

async def consumer(recv_channel, cancel_scope):
    async with recv_channel:
        async for item in recv_channel:
            if item == 42:
                cancel_scope.cancel()   # propagate shutdown
                break

async def main():
    send, recv = trio.open_memory_channel(100)
    async with trio.open_nursery() as nursery:
        nursery.start_soon(producer, send)
        nursery.start_soon(consumer, recv, nursery.cancel_scope)

trio.run(main)

For async generators: remember agen.aclose() if you stop early. Trio/AnyIO will help, but explicit closure makes tests reliable.

6) Threads — But Bounded and Explicit

Why it matters: Sometimes you can’t avoid blocking code (C extensions, legacy clients). Use thread offloading deliberately.

Trio

import trio, requests

def blocking_get(u):
    return requests.get(u, timeout=5).status_code

async def fetch(u):
    return await trio.to_thread.run_sync(blocking_get, u, cancellable=True)

AnyIO

import anyio

def heavy_cpu(x):  # or a blocking call
    return x ** 2

async def main():
    result = await anyio.to_thread.run_sync(heavy_cpu, 123)

Guardrails:

• Put a CapacityLimiter on thread usage.
• Mark the boundaries in code reviews (“leaves async world here”).
• If CPU-bound and continuous, consider a worker pool or process pool.

7) Cross-Runtime Portability with AnyIO (and When to Pick Trio)

Why it matters: Libraries shouldn’t force your runtime. AnyIO offers a compat layer so your code runs atop asyncio or Trio with minimal friction.

# entrypoint that can run on Trio or asyncio
import anyio

async def app():
    async with anyio.create_task_group() as tg:
        ...
anyio.run(app)      # Trio by default if installed; asyncio otherwise

When I pick Trio for services:

• You want structured concurrency and cancellation that feel ergonomic.
• You like nurseries and the mental model they enforce.
• You prefer explicitness over global event loops.

When I lean on AnyIO:

• I’m building a library consumed in mixed environments.
• I need a portable task group, capacity limiter, and fail_after without choosing a runtime for my users.

A Short Field Story

We replaced a homegrown asyncio crawler that occasionally OOM-ed under spikes. Two changes: (1) AnyIO task groups + capacity limiter at 50; (2) pipelines rewritten as async generators feeding a batched writer. Memory settled to a flat line, 95th percentile latency improved ~28%, and deploys got less scary. No exotic trickery — just bounded, structured, readable.

Testing Notes That Save Weekends

• Use a simulated clock (Trio has one) to make timeouts deterministic.
• Patch network with a fake httpx transport to avoid flaky tests.
• Probe cancellation: write one test that cancels mid-pipeline and asserts all channels/generators close.
• Property tests for generators: for any input stream of length N, the sum of emitted batch sizes equals N.

Common Pitfalls (and quick fixes)

• “It hangs on shutdown.” You forgot to propagate cancellation or close a channel/generator. Add explicit cancel_scope.cancel() or agen.aclose().
• “CPU spikes with 0 I/O.” You leaked an unbounded loop without await. Add await anyio.sleep(0) or restructure to an async generator that yields control.
• “Works locally, stalls in prod.” A hidden blocking call is sitting in the event loop. Wrap in to_thread.run_sync and cap with a limiter.

Wrap-Up

Concurrency that ships is rarely clever. It’s a handful of plays you trust: task groups, deadlines, capacity limiters, async generators, graceful shutdown, bounded threads, and portability with AnyIO. Start with one area — usually a streaming path or an outbound client — apply the pattern, and watch the tail latencies calm down.

If this was useful, drop a comment with the play you’ll implement first — or the one that burned you last time. I’ll share a small reference repo if there’s interest.