Async Generators

Async generators combine yield with async def to produce values lazily while awaiting I/O between them, and you consume them with async for.

An async generator is a function defined with async def that contains at least one yield statement. It behaves like a regular generator — calling it doesn't run its body, it returns an async generator object that produces values on demand — except that its body can also await other awaitables between yields. This makes async generators the natural tool for streaming data that arrives over time from an I/O source: paginated API results, chunks from a network socket, rows from an async database cursor.

What Is an Async Generator

A function is an async generator function if its definition starts with async def and its body contains yield (plain yield, not yield from, which is not permitted in async generators). Calling it returns an async generator object immediately — no code in the body runs yet. Values come out one at a time as the object is driven forward, and each step may suspend on an await before producing the next value.

Calling countdown(3) builds the async generator object but runs nothing; the loop below drives it forward.

import asyncio

async def countdown(n: int):
    while n > 0:
        yield n
        n -= 1
        await asyncio.sleep(0.1)   # simulate async work between values

async def main():
    agen = countdown(3)
    print(type(agen))   # <class 'async_generator'>
    async for value in agen:
        print(value)

asyncio.run(main())

async for and the Async Iterator Protocol

Async generators implement the asynchronous iterator protocol: __aiter__ returns the object itself, and __anext__ returns an awaitable that resolves to the next yielded value or raises StopAsyncIteration when the generator is exhausted. async for is the syntax that drives this protocol — it repeatedly awaits __anext__ and stops cleanly on StopAsyncIteration, the async counterpart of plain for calling next() until StopIteration.

You can drive an async generator manually with anext() (built-in since Python 3.10) or await agen.__anext__(), but async for is idiomatic for consuming the whole stream.

import asyncio

async def two_values():
    yield "first"
    yield "second"

async def main():
    agen = two_values()
    print(await anext(agen))
    print(await anext(agen))
    try:
        await anext(agen)
    except StopAsyncIteration:
        print("exhausted")

asyncio.run(main())

Generator vs Coroutine vs Async Generator

These three are easy to confuse because they all use async/await/yield in overlapping ways.

An async generator is a coroutine-like object that yields many times instead of returning once.

A function that is both async def and contains yield cannot also use a top-level return value — like sync generators, return in a generator (async or not) may only signal 'stop', not carry a value out through iteration.

Kind                | Defined with        | Produces         | Consumed with
---------------------|----------------------|------------------|------------------
plain generator      | def ... yield        | many values, lazily, sync | for / next()
coroutine             | async def (no yield) | one final result | await
async generator       | async def ... yield  | many values, lazily, async | async for / anext()

Streaming Data: Simulated Paginated Fetch

The core use case: producing items as they become available from an I/O source, without holding the whole sequence in memory or blocking the event loop while waiting for the next page.

Each page is awaited only when the consumer asks for more — items are streamed lazily rather than all fetched up front.

import asyncio

async def fake_fetch_page(page: int) -> list[int]:
    await asyncio.sleep(0.05)          # simulated network latency
    if page > 3:
        return []
    return [page * 10 + i for i in range(3)]

async def stream_all_items():
    page = 1
    while True:
        items = await fake_fetch_page(page)
        if not items:
            return
        for item in items:
            yield item
        page += 1

async def main():
    async for item in stream_all_items():
        print("received:", item)

asyncio.run(main())

Async Generator Expressions

Just as (x for x in it) is a generator expression, (x async for x in aiterable) is an async generator expression — usable anywhere an expression is valid, including as an argument. It requires an async context (inside an async def, or at the top level of certain async contexts).

import asyncio

async def numbers():
    for i in range(5):
        await asyncio.sleep(0)
        yield i

async def main():
    doubled = (n * 2 async for n in numbers() if n % 2 == 0)
    async for value in doubled:
        print(value)

asyncio.run(main())

Closing an Async Generator

Async generators support .aclose(), the async counterpart of a sync generator's .close(). It throws GeneratorExit into the generator at its current suspension point, letting any finally blocks or async with context managers inside it clean up (closing a file handle, a connection). If you abandon an async generator without exhausting or closing it, finally blocks in it may run later (or not at all, in some interpreter shutdown scenarios), so explicit cleanup matters for real resources.

import asyncio

async def resource_stream():
    print("opening resource")
    try:
        for i in range(100):
            yield i
    finally:
        print("closing resource")

async def main():
    agen = resource_stream()
    print(await anext(agen))
    print(await anext(agen))
    await agen.aclose()   # runs the finally block now, not later

asyncio.run(main())

Exception Handling in Async Generators

Exceptions raised inside an async generator's body propagate out through the await/__anext__ call that was driving it, exactly like a sync generator propagating out through next(). A try/except around the consuming async for loop (or around individual anext() calls) catches them the normal way.

import asyncio

async def risky_stream():
    yield 1
    yield 2
    raise ValueError("stream failed")

async def main():
    try:
        async for value in risky_stream():
            print(value)
    except ValueError as exc:
        print("caught:", exc)

asyncio.run(main())

Practical Usage Patterns

Real-world async generators typically wrap an I/O source: reading lines from an async file API, iterating rows from an async database driver (third-party libraries like asyncpg or motor expose async cursors this way), consuming messages from a websocket, or paginating through a REST API. Async generators are also the natural building block for a contextlib.asynccontextmanager-style pipeline stage that both manages a resource and streams data from it.

Async generators compose: one can consume another with async for and re-yield, building a pipeline.

async def filter_stream(source, predicate):
    """A reusable async generator that filters another async iterable."""
    async for item in source:
        if predicate(item):
            yield item

async def main():
    async for even in filter_stream(numbers(), lambda n: n % 2 == 0):
        print(even)

Edge Cases

An async generator cannot use yield from — there is no async equivalent of that sub-generator delegation syntax; instead you write an explicit async for ... yield loop to delegate to another async iterable.

Concurrently calling anext() on the *same* async generator object from two tasks at once is undefined/unsafe — an async generator, like a sync generator, is a single-threaded-at-a-time state machine; only one 'next step' should be in flight at a time.

An async generator that is garbage-collected without being closed may trigger a RuntimeWarning about the coroutine/generator never being awaited/closed, and its finally cleanup runs at an unpredictable time (or not at all) — always prefer explicit .aclose() or exhausting it fully.

Best Practices

Use async for to consume, never mix in a plain for (it will immediately raise TypeError: 'async_generator' object is not iterable).

Wrap resource-owning async generators in try/finally (or an inner async with) so cleanup runs even if the consumer stops early or an exception propagates.

Prefer an async generator over manually building and returning a list of awaited results when the sequence can be large or the source is naturally incremental — it keeps memory bounded and lets the consumer start processing before the whole sequence exists.

Don't do CPU-heavy synchronous work inside an async generator's body without yielding control — see the related Async Pitfalls topic on blocking the event loop.

Complete Example

Breaking out of an async for over an async generator triggers its cleanup, closing the underlying connection early.

import asyncio

async def fetch_lines(n: int):
    """Simulates streaming n lines from a network source."""
    try:
        for i in range(1, n + 1):
            await asyncio.sleep(0.02)
            yield f"line {i}"
    finally:
        print("connection closed")

async def process_lines(limit: int):
    count = 0
    async for line in fetch_lines(10):
        print("processing:", line)
        count += 1
        if count >= limit:
            print(f"stopping early after {limit} lines")
            break   # generator's aclose() runs via async for's implicit cleanup

asyncio.run(process_lines(4))

Mental Model

Think of an async generator as a paused coroutine that hands you one value at a time and remembers exactly where it left off — including which await it was mid-flight on. async for is a loop that keeps saying 'give me the next one, I'll wait' until the generator says 'no more.' The laziness of a plain generator and the non-blocking waiting of await are the same mechanism applied together.

Quick Reference

async def gen():         # async generator function
    yield value           # yields inside async def
    await something()     # can await between yields

async for x in gen():    # consume
    ...

agen = gen()
await anext(agen)         # manual single step (3.10+)
await agen.aclose()       # explicit cleanup / triggers finally

Key Takeaways

An async generator is async def plus yield; calling it returns an async generator object without running the body.

Consume with async for (or anext() for a single step); it implements __aiter__/__anext__ and raises StopAsyncIteration when done.

Async generators can await between yields, making them ideal for streaming data from I/O sources without blocking or buffering everything in memory.

yield from is not allowed inside an async generator; delegate with an explicit async for loop instead.

Call .aclose() (or let async for clean up on early exit) so finally blocks release resources promptly.

async def gen(): yield 1 yield 2 for x in gen(): # TypeError print(x) async def gen(): yield 1 yield 2 async def main(): async for x in gen(): print(x)

An async generator object is not a plain iterable — plain for raises TypeError; it must be driven with async for inside an async context.

async def gen(): for chunk in some_sync_generator(): yield chunk async def gen(): async for chunk in some_async_source(): yield chunk

Delegating to another generator inside an async generator should use async for over an async source, not a plain for — mixing sync iteration in is fine only when the source is genuinely synchronous and non-blocking.

agen = stream() # consumer stops early, no cleanup, connection leaks first = await anext(agen) del agen agen = stream() try: first = await anext(agen) finally: await agen.aclose()

Abandoning an async generator without closing it leaves any finally/async with cleanup inside it to run at an unpredictable time (or not at all).

What does calling an async generator function return, and when does its body start executing?

It returns an async generator object immediately; no body code runs until the object is driven with async for, anext(), or .asend().

Why can't yield from be used inside an async generator?

There's no async version of sub-generator delegation syntax; you must write an explicit async for ... yield loop to delegate to another async iterable.

What exception does async for catch internally to know an async generator is exhausted?

StopAsyncIteration, raised when __anext__ has no more values to produce.

Why is an async generator a good fit for a paginated API client?

It can await each page's network fetch lazily, yielding items as they arrive, without loading every page into memory up front.

What happens if you use plain for on an async generator?

TypeError, because an async generator object implements __aiter__/__anext__, not the synchronous __iter__/__next__ protocol.

How do you manually pull one value from an async generator without a loop?

await anext(agen) (Python 3.10+), or await agen.__anext__() directly.

Why should resource-owning async generators use try/finally?

So that breaking out of the consuming loop or calling .aclose() still runs cleanup code, releasing connections or files promptly.

Write an async generator evens_up_to(n) that awaits asyncio.sleep(0) before yielding each even number from 0 up to and including n.

import asyncio async def evens_up_to(n: int): for i in range(0, n + 1, 2): await asyncio.sleep(0) yield i async def main(): async for x in evens_up_to(10): print(x) asyncio.run(main())

Prints 0, 2, 4, 6, 8, 10 on separate lines, each after yielding control back to the event loop via asyncio.sleep(0).

What happens when this runs?

It prints 1, then 2, then raises StopAsyncIteration on the third anext() call because the generator is exhausted — the program crashes with that unhandled exception.

Write a coroutine that consumes an async generator of numbers and returns their sum, stopping early (and closing the generator) if a running total exceeds 100.

async def sum_until_limit(agen, limit: int = 100) -> int: total = 0 try: async for value in agen: total += value if total > limit: break finally: await agen.aclose() return total

Iterates the async generator, accumulating total; once it exceeds limit it breaks out of the loop, and the finally block ensures .aclose() runs regardless of how the loop ended.

async def gen():
    yield 1
    yield 2

async def main():
    agen = gen()
    print(await anext(agen))
    print(await anext(agen))
    print(await anext(agen))

import asyncio
asyncio.run(main())