Async iterators and async generators

Everything in the synchronous iteration protocol has an async twin, and one extra problem that has no synchronous equivalent: you cannot clean up an async generator without an event loop, so its cleanup cannot happen at the moment the collector notices it is garbage.

That single fact is why async iteration needs a discipline that sync iteration does not.

The protocol

class Reader:
    def __aiter__(self) -> Self:          # NOT async def
        return self

    async def __anext__(self) -> bytes:   # IS async def
        line = await self._readline()
        if not line:
            raise StopAsyncIteration
        return line

Three rules:

1. __aiter__ must not be async def. It is a plain method that returns the asynchronous iterator. Before 3.7, __aiter__ was allowed to return an awaitable resolving to an async iterator; since 3.7 it must return the iterator itself, and returning anything else raises TypeError. Writing async def __aiter__ today means every async for over your object fails at the first iteration.
2. __anext__ must return an awaitable, so it is async def. It raises StopAsyncIteration — not StopIteration — to end the stream.
3. async for is what drives both, and it can only appear inside an async def.

Since 3.10 there are builtins for driving the protocol by hand: aiter(obj) and anext(it), plus the two-argument anext(it, default), which returns default instead of raising StopAsyncIteration. They are the async analogues of iter and next, including the sentinel form.

Async generators

Writing the class by hand is rarely necessary. An async def containing yield is an async generator function:

from collections.abc import AsyncGenerator


async def read_pages(client: Client) -> AsyncGenerator[Page, None]:
    cursor: str | None = None
    while True:
        batch = await client.fetch(cursor)
        if not batch.items:
            return
        for page in batch.items:
            yield page
        cursor = batch.next_cursor

Note the return annotation has two parameters — AsyncGenerator[Yield, Send]. There is no third, because an async generator cannot return a value; return inside one only ends the stream.

As with sync generators, calling read_pages(client) executes none of the body. It builds the async generator object and returns it. The await client.fetch(...) has not happened.

💡A function annotated async def f(...) -> AsyncIterator[T] — is f(...) an async iterator, or a coroutine that produces one? How can you tell without running it? click to reveal

It depends entirely on whether the body contains a yield, and that is the whole gotcha.

• No yield in the body: f is an ordinary coroutine function. f(...) returns a coroutine, and you must await f(...) to get the AsyncIterator[T]. Its real type is Coroutine[Any, Any, AsyncIterator[T]].
• Any yield in the body: f is an async generator function. f(...) returns the async iterator directly, and awaiting it is a TypeError.

The annotation is identical in both cases. Adding or deleting one yield during a refactor silently flips the calling convention of your function, and mypy follows along without complaint because both are legal readings of that signature.

Two practical consequences. First, reveal_type(f(...)) is how you check — mypy prints Coroutine[...] or AsyncIterator[...] and settles it. Second, an async generator cannot validate its arguments at call time, because none of its body runs until the first __anext__. If you need eager validation, split into a plain def wrapper that checks and returns, plus the async def ... yield worker. That split is not a style preference; it is the only way to raise before the caller starts iterating.

The finalisation problem

Here is the code that looks fine and is not:

async def process(client: Client) -> None:
    async for page in read_pages(client):
        if page.is_terminal:
            break            # <- read_pages is now suspended, forever-ish

When you break, the async generator is left parked at its yield, holding whatever it was holding — an open connection, a cursor on the server, a semaphore slot. Its finally block has not run.

For a synchronous generator, CPython’s collector would call close() during finalisation and the finally would run more or less immediately. An async generator cannot be finalised that way, because closing it may need to await — and there is no event loop inside __del__. PEP 525 solved this with sys.set_asyncgen_hooks, which asyncio uses to register every async generator that starts iterating, and asyncio.run calls loop.shutdown_asyncgens() on the way out.

So the cleanup does eventually happen — at loop shutdown. Which means:

• In a script, your finally runs when asyncio.run returns. Possibly seconds late, but it runs.
• In a long-lived service, the loop shuts down when the process does. Your connection is held until then.
• The exception traceback, if the finally raises, arrives detached from the code that caused it.

“Cleanup happens at process exit” is not resource management.

The fix is two lines

contextlib.aclosing (3.10) makes it deterministic:

from contextlib import aclosing


async def process(client: Client) -> None:
    async with aclosing(read_pages(client)) as pages:
        async for page in pages:
            if page.is_terminal:
                break
    # aclose() has already been awaited here. The finally has already run.

aclosing.__aexit__ awaits pages.aclose(), which throws GeneratorExit into the generator at its suspended yield. The generator’s finally runs, awaiting whatever it needs to, and the whole thing completes before the async with block exits. Synchronously with respect to your code.

aclosing needs nothing but an aclose() method, so it works on hand-written async iterators too — which is a good reason to give your async iterator classes an aclose() even when they wrap something that has no obvious close.

💡Inside an async generator's finally, which of these is safe: await something(), yield x, async with open_scope():? click to reveal

await something() is safe. Closing an async generator drives its frame to completion, and awaiting is exactly how it gets there — this is the entire reason aclose() is a coroutine rather than a method.

yield x is not safe and is a hard error. Yielding while handling GeneratorExit means refusing to close; the interpreter raises RuntimeError: async generator ignored GeneratorExit. If you find yourself wanting to emit one last item on the way out, redesign — the consumer has already stopped listening.

async with open_scope(): depends on what the scope is. An ordinary async context manager is fine. A cancel scope — a TaskGroup or asyncio.timeout — is a different matter, and opening one during finalisation is asking for a cancellation to be delivered to whichever task happens to be driving aclose(). That is the subject of PEP 789, and the review rule is: no cancel scopes across a yield, and none in a generator’s cleanup path.

There is also a timing constraint people miss: cleanup during loop shutdown runs with limited time and, if the loop is already closing, an await on a network round trip may never complete. Keep async finally blocks to local operations — release a semaphore, return a connection to a pool, cancel a task — not to work that needs a live socket.

Async comprehensions and async for in expressions

names = [user.name async for user in fetch_users()]
live = {u.id: u async for u in fetch_users() if await u.is_active()}

Both are ordinary comprehensions with async for, both only legal inside an async def, and both materialise — the first line holds every user in memory. The lazy equivalent is a generator expression with async for, which produces an async generator:

names = (user.name async for user in fetch_users())    # AsyncGenerator[str, None]

and which, being an async generator, brings the finalisation problem with it. If you break out of a loop over one of those, wrap it in aclosing like any other.

A checklist

• __aiter__ is def, __anext__ is async def. Not the other way round.
• Annotate async generators AsyncGenerator[Yield, Send], or AsyncIterator[T] if nobody sends.
• Any async for you might break out of: wrap the iterator in aclosing.
• Validate eagerly in a plain def wrapper; an async generator cannot raise at call time.
• Never yield inside a TaskGroup or asyncio.timeout scope.
• Give hand-written async iterators an aclose(), so callers can use the same discipline.