Keep yield Outside Timeouts and TaskGroups

You give each read from a network source a time limit, inside an async generator. The reads are quick. The caller then spends 0.2 seconds saving each item, and the program crashes:

import asyncio
from collections.abc import AsyncIterator


async def read_items() -> AsyncIterator[int]:
    for n in range(3):
        async with asyncio.timeout(0.1):  # each read gets 0.1 seconds
            await asyncio.sleep(0.01)  # stands in for a quick read
            yield n  # bug: the timer keeps running while the caller works


async def main() -> None:
    async for n in read_items():
        print("saving", n)
        await asyncio.sleep(0.2)  # the caller's own work
        print("saved", n)


asyncio.run(main())

saving 0
Traceback (most recent call last):
  ...
asyncio.exceptions.CancelledError

The program dies before it prints saved 0. The timer in read_items cancelled the caller's own asyncio.sleep(0.2).

A paused generator leaves its blocks open

As the cancellation article showed, asyncio.timeout cancels the code in its block when time runs out. It does that by cancelling the task that entered the block. When the generator reaches yield n, it pauses with the async with still open. The same task goes on to run the caller's loop body, and the timer keeps running. When its 0.1 seconds run out, it cancels the task, and the caller's asyncio.sleep(0.2) raises CancelledError.

The error goes up through the caller's code and out of main. The timeout block is in the paused generator, out of the error's path, so the caller gets a bare CancelledError that nothing catches. The TimeoutError shows up only in a log line while asyncio.run shuts down.

A cancel scope is a block that can cancel the task running it. In asyncio, async with asyncio.timeout(...) and async with asyncio.TaskGroup() both open one. The rule is the same for both. In an async generator, end the block before you yield.

End the timeout before you yield

Move the yield out of the async with:

async def read_items() -> AsyncIterator[int]:
    for n in range(3):
        async with asyncio.timeout(0.1):  # each read gets 0.1 seconds
            await asyncio.sleep(0.01)  # stands in for a quick read
        yield n  # the timer has stopped before the caller gets the item

Now the timer covers only the read, and the caller can take as long as it needs with each item. Each read gets a fresh 0.1 seconds. If you want one limit for the whole job, put the timeout around the caller's loop instead.

Keep a TaskGroup open around the caller's block, not across a yield

A TaskGroup has the same problem. When one of its tasks fails, the group cancels the task that opened it. That is how it stops the code in its block before it raises the error. While the generator stays paused at a yield, the caller's code gets that cancellation:

async def fail_soon() -> None:
    await asyncio.sleep(0.05)
    raise ValueError("background fetch failed")


async def pages() -> AsyncIterator[int]:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fail_soon())
        for n in range(3):
            yield n  # bug: the group stays open while the caller works

A caller that loops over pages() the way main did gets a CancelledError in the middle of its save. The ValueError goes to a log line at shutdown, where no handler can catch it.

When a stream needs background tasks, let the caller's block own the group. @asynccontextmanager, from contextlib, turns an async generator that yields once into a context manager that you use with async with. The code before its yield runs when the block starts, and the code after runs when the block ends.

import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager


async def fail_soon() -> None:
    await asyncio.sleep(0.05)
    raise ValueError("background fetch failed")


async def numbers() -> AsyncIterator[int]:
    for n in range(3):
        yield n


@asynccontextmanager
async def open_pages() -> AsyncIterator[AsyncIterator[int]]:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(fail_soon())
        yield numbers()  # one yield, and the caller's whole block runs here


async def main() -> None:
    try:
        async with open_pages() as pages:
            async for n in pages:
                print("saving", n)
                await asyncio.sleep(0.2)
                print("saved", n)
    except* ValueError as group:
        print("caught:", group.exceptions)


asyncio.run(main())

saving 0
caught: (ValueError('background fetch failed'),)

This yield is the context manager's single yield, and the caller's whole async with block runs inside the group on purpose. When fail_soon raises, the group cancels the block and waits for its tasks. Then it raises the ExceptionGroup at the end of the async with, where the caller catches it. The stream itself, numbers(), yields outside any cancel scope of its own.

Common mistakes

• Yielding stream items inside async with helper():, where helper opens a timeout or TaskGroup. The rule covers every block that can cancel, whoever wrote it.
• Treating aclosing as a cancel scope. It only awaits aclose() when its block ends. It sets no timer and cancels nothing, so a yield inside async with aclosing(...) is fine.

Check your understanding Your async generator wraps both the read and the yield in async with asyncio.timeout(1):. The caller spends two seconds saving each item. What happens? Show answer Hide answer

After one second, the timer cancels the task while the caller is saving. The generator stays paused at its yield, and its timeout block is out of the error's path. So the caller gets a bare CancelledError instead of a TimeoutError. End the async with block before the yield.

Going further

PEP 789 describes these bugs and proposes a way for Python to reject such yields. It also points to the ASYNC101 rule of the flake8-async linter, which catches the common cases.