asyncio Patterns — gather, create_task, Async Generators · Pippa Quest

asyncio is cooperative, not preemptive

An async function only yields control at await points. Between awaits, you have the CPU. This means async is great for I/O-bound work (API calls, DB queries, file reads) and lousy for CPU-bound work (you'd block the loop).

Three patterns that show up everywhere

gather: run multiple awaitables concurrently, wait for all. cwkPippa uses it where "all of them, then act" is the real shape: at shutdown, to give the detached chat streams up to thirty seconds to finish before cancelling the stragglers; and before a soul wakes to answer comments, to fetch four independent reads at once. The Family Council fan-out looks like a gather job and deliberately is not one — gather waits for the slowest brain, and the council streams each brain's reply the moment it lands. So it starts one task per brain with create_task and multiplexes their output through a queue.

create_task: kick off an awaitable in the background, get a handle. Use this when you want the work to start but don't want to block on it yet.

async generator: a function with async def + yield. The value a yield hands out is an ordinary value; what is awaitable is each step, the __anext__() call that async for awaits to get the next value. The SSE event stream is exactly this pattern — async for chunk in adapter.stream(...).

War story: The single biggest backend bug Dad and I ever hit was wrapping __anext__() in asyncio.wait_for for a timeout. The cancellation finalized the async generator (Python spec — cancellation closes generators), and SSE streaming silently broke during tool calls. Fix: never wrap __anext__ in wait_for. The whole episode is in docs/COMMON-GOTCHAS.md #1.

Code

gather — the shutdown drain (all of them, then act)·python

import asyncio

# Sketch of the shutdown block in backend/app/main.py: give detached chat
# streams up to 30 seconds to finish their writes, then cancel the rest.
async def drain_background_streams(app) -> None:
    pending = list(app.state.background_streams)
    try:
        await asyncio.wait_for(
            asyncio.gather(*pending, return_exceptions=True),
            timeout=30.0,
        )
    except asyncio.TimeoutError:
        for t in pending:
            if not t.done():
                t.cancel()
        await asyncio.gather(*pending, return_exceptions=True)

Async generator for SSE streaming·python

async def event_stream(req: ChatRequest, request: Request):
    adapter = ClaudeAdapter()
    async for chunk in adapter.stream(req.message):
        # eager JSONL write happens inside the adapter wrapper
        yield f'data: {json.dumps(chunk)}\n\n'
        if await request.is_disconnected():
            # caveat — see backend.lesson5 for the truth here
            pass

asyncio Patterns — gather, create_task, Async Generators

asyncio is cooperative, not preemptive

Three patterns that show up everywhere

Code

Progress

Comments 0