Async Iterators, Generators and Context Managers

New words in this topic, in plain English. Come back here whenever one feels fuzzy.

Async iterator

An object you loop over with async for. Getting its next value can await.

Async generator

An async def function with yield that produces values over time.

async for

A loop that awaits each next value of an async iterator.

async with

A with block whose entry and exit can await.

Streaming

Receiving a result in small pieces as it is produced, instead of all at the end.

Back-pressure

A slow consumer making a fast producer wait, so memory does not fill up.

aclose

The method that shuts down a half-finished async generator.

Sentinel

A special value, like None, put in a queue to signal that no more work is coming.

1. 1

   Break at a different token

   In the first example change "three" to "two" and predict what stopped after: prints. Then remove await stream.aclose() and run again. The output looks the same; the difference is in cleanup that you do not see. Add a try/finally with a print inside stream_tokens to see when cleanup runs.

2. 2

   Change the queue size

   In the queue example use Queue(maxsize=1), then Queue() (unlimited). The results stay the same. What changes is how far ahead the producer can run, which matters when producing is cheap and consuming is slow.

3. 3

   Hide an exception

   In the Connection example change return False to return True in __aexit__. The ValueError is now swallowed and the last print never runs. Hiding errors is almost always a bug; do it only on purpose.

Errors are the best teachers. Make each change, read the error, guess what went wrong, then reveal the answer.

import asyncio


async def tokens():
    for word in ["a", "b"]:
        await asyncio.sleep(0.01)
        yield word


async def main():
    for token in tokens():
        print(token)


asyncio.run(main())

1. 1

   A normal for loop calls next() and gets the next value at once (Topic 9.1). An async iterator has the special methods __aiter__ and an async __anext__, so getting the next value can await (for a network chunk, a database row, a token from a model). async for item in source: calls await source.__anext__() each round and stops when StopAsyncIteration is raised. It works only inside async def.

2. 2

   An async generator is the easy way to build one: an async def function that contains yield. Each yield hands one value to the async for loop, and the function can await between yields. It is the async version of the generators in Topic 9.2. Unlike a coroutine, calling it does not need await: it returns an async generator object you loop over with async for.

3. 3

   Streaming LLM output is exactly this. A model API sends the answer as many small pieces as they are produced. A piece is a token (a chunk of text, roughly a word or part of a word, Topic 16.4) or a larger chunk. Your code consumes them with async for chunk in stream: and prints or forwards each one immediately, so the user sees the first words at once instead of waiting for the whole reply. The example uses a fake model with asyncio.sleep; real SDKs offer the same shape.

4. 4

   Stopping early. If the consumer breaks out of the loop, the generator is left suspended. Close it with await gen.aclose() (or wrap it in contextlib.aclosing(...)) so its finally blocks and any open connection are cleaned up. A generator that is just dropped is cleaned up later by the event loop, at a time you do not control.

5. 5

   **async with** works like with (Topic 7.4), but entering and leaving can await: __aenter__ and __aexit__ are coroutines. Use it for things that open and close over the network: a database connection, an HTTP client, a lock, a semaphore, a TaskGroup, a timeout. You can write one as a class with the two methods, or as a generator decorated with @contextlib.asynccontextmanager. Returning True from __aexit__ hides an exception; return False (or nothing) to let it through.

6. 6

   **asyncio.Queue** is the async version of a thread queue (Topic 12.3): await queue.put(item) waits if the queue is full (back-pressure), await queue.get() waits if it is empty, task_done() marks an item handled, and await queue.join() waits for all items. A common design is some producer tasks, a queue, and a few worker tasks; the queue's size limit stops fast producers from overloading slow consumers. Async comprehensions ([x async for x in gen]) collect a whole stream into a list.

• ↔

  Streaming vs waiting for the whole result: streaming shows the first piece sooner and uses less memory, but your code must handle partial results and errors in the middle of a stream.

• ↔

  Bounded vs unbounded queue: a bound protects memory and gives back-pressure, but can slow producers and causes deadlock if a consumer never takes items.

• ↔

  Class vs @asynccontextmanager: the decorator is shorter; the class is clearer when you need state or want to control exactly what __aexit__ does with exceptions.

• Done when you can write an async generator and consume it with async for.

• Done when you can explain why a streaming LLM response is an async iterator.

• Done when you can stop a stream early and close it properly.

• Done when you can write an async with manager in both forms.

• Done when you can build a producer, a bounded queue and workers, and shut them down cleanly.