C.W.K.
Stream
Lesson 02 of 05 · published

Async I/O Patterns — Streaming, Timeouts, Cancellation

~20 min · async, stream, timeout, cancellation

Level 0Curious
0 XP0/93 lessons0/23 achievements
0/100 XP to next level100 XP to go0% complete

Async iteration — async for, async generators

Just as for x in seq works on iterables, async for x in stream works on async iterables. An async generator is just a function with both async def and yield — each yield can be preceded by await. This is how cwkPippa's chat handlers work: yield each chunk of the response as it streams in.

asyncio.timeout — the modern way

3.11+ introduced async with asyncio.timeout(seconds): as the modern timeout pattern. Pre-3.11, you'd use asyncio.wait_for(coro, timeout). Both raise TimeoutError if the body doesn't finish in time. The timeout context manager composes better with other async code.

Cancellation — when something needs to stop

A task can be cancelled with task.cancel(). Inside the task, this raises CancelledError at the next await point. You can catch it for cleanup but should generally re-raise. CancelledError isn't a regular exception — try/except Exception doesn't catch it. This is intentional; cancellation should not be silently swallowed.

shield — protecting from cancellation

asyncio.shield(coro) protects an inner coroutine from cancellation propagating from the outside. The outer task gets cancelled, but the shielded inner one keeps running. Useful for "please finish writing this critical record before we abort."

War Story: One of cwkPippa's most-cited gotchas: never wrap __anext__ in asyncio.wait_for. The cancellation finalizes the async generator, and any subsequent next call raises StopIteration instead of recovering. The right shape: do timeout management around the whole streaming operation, not around each individual chunk.

Code

async for over an async generator·python
import asyncio

async def number_stream(start, end, delay=0.1):
    for i in range(start, end):
        await asyncio.sleep(delay)
        yield i

async def main():
    async for n in number_stream(0, 5):
        print(n)

asyncio.run(main())
# 0, 1, 2, 3, 4 — one every 100ms
Timeout — the 3.11+ pattern·python
import asyncio

async def slow_op():
    await asyncio.sleep(5)
    return "done"

async def main():
    try:
        async with asyncio.timeout(1):
            result = await slow_op()
            print(result)
    except TimeoutError:
        print("timed out")

asyncio.run(main())

# Pre-3.11 way (still works):
async def main_old():
    try:
        result = await asyncio.wait_for(slow_op(), timeout=1)
        print(result)
    except TimeoutError:
        print("timed out (old style)")

asyncio.run(main_old())
Cancellation — catch and re-raise for cleanup·python
import asyncio

async def long_task():
    try:
        for i in range(100):
            print("working", i)
            await asyncio.sleep(0.1)
    except asyncio.CancelledError:
        print("got cancellation — cleaning up")
        # cleanup work here
        raise                               # re-raise so cancellation propagates

async def main():
    task = asyncio.create_task(long_task())
    await asyncio.sleep(0.3)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("task confirmed cancelled")

asyncio.run(main())
shield — keep something running despite cancellation·python
import asyncio

async def critical_save():
    print("saving...")
    await asyncio.sleep(0.5)
    print("saved")
    return "checkpoint"

async def main():
    task = asyncio.create_task(critical_save())
    try:
        # Even if outer is cancelled, the shielded task continues
        result = await asyncio.shield(task)
        print("result:", result)
    except asyncio.CancelledError:
        print("outer cancelled, but inner task is still running")
        # Wait for the protected work to actually finish
        result = await task
        print("after wait:", result)

asyncio.run(main())
asyncio.Queue — async producer/consumer·python
import asyncio

async def producer(queue, count):
    for i in range(count):
        await asyncio.sleep(0.1)
        await queue.put(f"item-{i}")
        print(f"put item-{i}")
    await queue.put(None)                   # signal done

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

async def main():
    queue = asyncio.Queue(maxsize=2)
    await asyncio.gather(producer(queue, 5), consumer(queue))

asyncio.run(main())

External links

Exercise

Write async def slow_fetch(name, delay) that sleeps for delay seconds and returns f"{name}-done". Run two of these concurrently with asyncio.gather, but wrap both in asyncio.timeout(1). Test with delays of 0.5 (both finish) and 2 (timeout fires). Then catch the TimeoutError and print which inputs caused it. Confirm cleanup happens correctly.

Progress

Progress is local-only — sign in to sync across devices.
Spotted a bug or have feedback on this page?Report an Issue

Comments 0

🔔 Reply notifications (sign in)
Sign inPlease sign in to comment.

No comments yet — be the first.