Stream Paged Results with Async Generators

Your app lists a customer's orders from an API that returns two at a time. You want to write async for order in orders(): and have the next page fetched only when the loop needs it. An async generator does this. It is an async def function that contains yield:

import asyncio
from collections.abc import AsyncIterator


async def fetch_page(number: int) -> list[str]:
    print(f"fetching page {number}")
    await asyncio.sleep(0.1)  # stands in for an API call
    return [f"order-{number}-{i}" for i in range(2)] if number < 2 else []


async def orders() -> AsyncIterator[str]:
    number = 0
    while True:
        page = await fetch_page(number)
        if not page:  # an empty page means there are no more
            return
        for order in page:
            yield order
        number += 1


async def main() -> None:
    async for order in orders():
        print(order)


asyncio.run(main())

fetching page 0
order-0-0
order-0-1
fetching page 1
order-1-0
order-1-1
fetching page 2

It works like the generators from the generators track, except that its body can await. Calling orders() runs nothing. It returns an async iterator, an object you loop over with async for. Each step of the loop runs the body up to the next yield.

Annotate it with the type of the items it yields, -> AsyncIterator[str]. You loop over an async generator. You can't await it.

async for waits for one item at a time

async for asks for an item, waits for it, runs the loop body, then asks for the next one. The output shows it. The generator fetches page 1 only after the loop has printed both orders from page 0. For paging, that is what you want, because you never fetch a page nobody reads. To fetch several pages at the same time, use gather, as in asyncio from Zero.

An async comprehension, [order async for order in orders()], reads the whole stream into a list.

Write the protocol by hand

async for works with any object that has two methods:

• __aiter__(), a plain def that returns the async iterator, usually self.
• __anext__(), an async def that returns the next item, or raises StopAsyncIteration when there are no more.

import asyncio
from typing import Self


class Countdown:
    def __init__(self, start: int) -> None:
        self.current = start

    def __aiter__(self) -> Self:
        return self

    async def __anext__(self) -> int:
        if self.current <= 0:
            raise StopAsyncIteration
        self.current -= 1
        return self.current + 1


async def main() -> None:
    print([n async for n in Countdown(3)])  # [3, 2, 1]
    numbers = aiter(Countdown(2))
    print(await anext(numbers))  # 2
    print(await anext(numbers))  # 1
    print(await anext(numbers, None))  # None, because the stream has ended


asyncio.run(main())

The built-in aiter() and anext() are the async versions of iter() and next(). anext() returns an awaitable, so you await it. Its second argument is a default it returns when the stream has ended.

Write a class when the iterator has state that other code needs to see, such as a page number or a closed flag. The exercise builds one.

Close a stream you leave early

When you break out of async for, the generator stays paused at its yield. Its finally blocks have not run, so a connection it holds is still open. The event loop closes the generator later, at a time you don't control. contextlib.aclosing closes it when your block ends:

import asyncio
from collections.abc import AsyncGenerator
from contextlib import aclosing


async def rows() -> AsyncGenerator[int, None]:
    print("connection opened")
    try:
        for n in range(3):
            await asyncio.sleep(0)  # stands in for a database read
            yield n
    finally:
        print("connection closed")


async def main() -> None:
    async with aclosing(rows()) as stream:
        async for row in stream:
            print("got", row)
            break
    print("after the block")


asyncio.run(main())

connection opened
got 0
connection closed
after the block

When the block ends, aclosing awaits the generator's aclose() method. That raises GeneratorExit at the paused yield, so the finally runs before your next line. aclosing works with any object that has an aclose() method, including a class you write. A later article covers it with the other cleanup helpers in contextlib.

For mypy, annotate a generator that callers will close as -> AsyncGenerator[int, None]. AsyncIterator[int] doesn't promise an aclose() method, so mypy rejects aclosing(...) on it. The None is the type of values sent into the generator, which you rarely use.

Common mistakes

• Writing async def __aiter__. Then __aiter__ returns a coroutine, and async for fails with this error: TypeError: 'async for' received an object from __aiter__ that does not implement __anext__: coroutine. Make __aiter__ a plain def.
• Forgetting await before anext(...). You get an awaitable instead of the next item.
• Checking an argument inside the async generator. As with any generator, the body runs only when the loop asks for the first item, so a bad argument fails later than the call. Check it in a plain function that then returns the generator.

Check your understanding You break out of async for row in rows(db): after the first row. Inside rows, an async with block holds a database connection. When does the connection go back? Show answer Hide answer

Not at the break. The generator stays paused at its yield, still inside the async with. The event loop closes it some time later, once nothing refers to it, or when asyncio.run shuts down. To return the connection when you stop reading, write async with aclosing(rows(db)) as stream: and loop over stream.

Going further

An async generator can stop with a bare return, but it can't return a value. Its finally can await, for example to close a connection. If it yields again while aclose() runs, Python raises RuntimeError.

See contextlib.aclosing and the aiter and anext built-ins.