Skip to content

Building Async Iterator Classes with aiter and anext

Async generators are the default way to write an async iterator, and usually the right one. A class implementing __aiter__ and __anext__ earns its place in three situations: the iterator has state callers need to inspect or change mid-iteration (a cursor position, a buffer to peek into, a "push this item back" operation), cleanup must happen at a moment you control rather than when a generator happens to be finalised, or the object should be both an iterator and something else — a stream with close(), a subscription with ack(). Performance is not the reason: in a microbenchmark a class iterator cost 107 ns per item and an equivalent async generator 111 ns. This guide builds a class iterator correctly and covers the builtins aiter() and anext() added in Python 3.10.

Prerequisites

1. Implement the protocol correctly

The protocol has two methods and one exception:

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

    def __aiter__(self) -> "Countdown":
        return self                                  # a plain method, NOT async def

    async def __anext__(self) -> int:
        if self._n <= 0:
            raise StopAsyncIteration                 # not StopIteration
        await asyncio.sleep(0)                       # any awaits go here
        self._n -= 1
        return self._n + 1

Two rules catch people. __aiter__ must be a regular method returning the iterator; declaring it async def makes async for fail with TypeError: 'async for' received an object from __aiter__ that does not implement __anext__: coroutine. And the end of iteration is signalled with StopAsyncIteration; raising StopIteration inside a coroutine is converted to RuntimeError by Python.

Returning self from __aiter__ makes the object a one-shot iterator: iterating it twice yields nothing the second time. If the object should be re-iterable — like a list — __aiter__ should return a fresh iterator object each time.

Verify: async for x in Countdown(3) yields 3, 2, 1, and a second loop over the same object yields nothing.

What async for actually calls A sequence of 6 messages between 2 participants. What async for actually calls async for iterator object __aiter__() → iterator (sync call) await __anext__() value 3 await __anext__() value 2 … value 1 raise StopAsyncIteration: loop ends __aiter__ is called once and synchronously; every item costs one awaited __anext__.

2. Use a class when callers need the state

The real reason to write a class is state that callers can see and influence. A paginated reader that exposes its cursor, supports peeking, and can push an item back for re-processing is awkward as a generator and natural as a class:

from collections import deque


class PagedReader:
    def __init__(self, client, path: str, page_size: int = 100) -> None:
        self._client, self._path, self._size = client, path, page_size
        self._buf: deque = deque()
        self.cursor: str | None = None               # visible: callers can checkpoint it
        self._done = False

    def __aiter__(self) -> "PagedReader":
        return self

    async def __anext__(self):
        if not self._buf:
            await self._fill()
        if not self._buf:
            raise StopAsyncIteration
        return self._buf.popleft()

    async def peek(self):
        if not self._buf:
            await self._fill()
        return self._buf[0] if self._buf else None

    def push_back(self, item) -> None:
        self._buf.appendleft(item)

    async def _fill(self) -> None:
        if self._done:
            return
        resp = await self._client.get(self._path, params={"cursor": self.cursor, "limit": self._size})
        body = resp.json()
        self._buf.extend(body["items"])
        self.cursor = body.get("next")
        self._done = self.cursor is None

A caller can save reader.cursor to resume after a crash, look ahead with peek() to decide whether to stop, or push_back() an item it could not handle yet. Doing the same with an async generator means smuggling state out through asend() or a shared object — possible, but the class is clearer. The generator version for simple cases is in writing async iterators for paginated APIs.

Verify: iterate halfway, save the cursor, build a new reader from it, and confirm it resumes at the next page.

3. Make cleanup explicit with aclose and async with

An async generator's finally block runs when the generator is closed — which, if the consumer breaks out of the loop without aclosing(), happens whenever the garbage collector and the loop's async-generator hooks get to it. A class has no implicit finaliser for async cleanup at all, which forces you to make cleanup explicit — a feature, not a limitation:

class Subscription:
    def __init__(self, conn, channel: str) -> None:
        self._conn, self._channel = conn, channel
        self._queue: asyncio.Queue = asyncio.Queue(maxsize=1000)
        self._closed = False

    async def __aenter__(self) -> "Subscription":
        await self._conn.subscribe(self._channel, self._queue.put_nowait)
        return self

    async def __aexit__(self, *exc) -> bool:
        await self.aclose()
        return False

    def __aiter__(self) -> "Subscription":
        return self

    async def __anext__(self):
        if self._closed:
            raise StopAsyncIteration
        return await self._queue.get()

    async def aclose(self) -> None:
        if not self._closed:
            self._closed = True
            await self._conn.unsubscribe(self._channel)


async with Subscription(conn, "orders") as sub:
    async for msg in sub:
        if msg.kind == "shutdown":
            break                                  # __aexit__ unsubscribes, right now

Pairing the iterator with __aenter__/__aexit__ means the resource lifetime is the async with block, independent of when the iteration stops or whether the object is garbage-collected.

Verify: break out of the loop; the server-side subscription count drops immediately, not at the next GC cycle.

Class iterator or async generator? A grid of 5 rows by 3 columns. Class iterator or async generator? property class with __anext__ async generator cost per item 107 ns 111 ns state visible to callers attributes and methods hidden in the frame peek, push back, checkpoint natural awkward cleanup timing explicit aclose or async with finaliser unless aclosing() code size larger smallest Choose by the interface you need; the speed difference is noise.

4. Use the aiter() and anext() builtins

Python 3.10 added aiter() and anext() as the async counterparts of iter() and next(). anext() is the useful one: it fetches a single item without a loop, and accepts a default for exhaustion:

async def first_or_none(source):
    return await anext(aiter(source), None)


async def header_and_rows(stream):
    it = aiter(stream)
    header = await anext(it)                     # raises StopAsyncIteration if empty
    async for row in it:                         # continue from where anext left off
        yield dict(zip(header, row))

Verified: after three items from a three-item generator, await anext(it, "done") returned "done" rather than raising. Without a default, anext() on an exhausted iterator raises StopAsyncIteration — outside an async for that is an ordinary exception you must catch.

aiter() on an object that only has __aiter__ returning a non-iterator raises TypeError immediately, which makes it a cheap way to validate inputs at a function boundary instead of failing on first iteration.

Verify: await anext(aiter([]))-style misuse fails fast with TypeError, because a list is not async-iterable.

Pulling items by hand with aiter and anext A flow of 4 stages. Pulling items by hand with aiter and anext aiter(source) validate, get iterator await anext(it) take the first item async for row in it continue from there anext(it, default) no exception at the end anext is how you take one item from an async iterator without writing a loop.

5. Keep anext cancellation-safe

__anext__ is a coroutine and can be cancelled at any await inside it. If it has partially updated state when cancelled — popped an item from the buffer but not returned it, advanced the cursor before the fetch finished — that item is lost:

async def __anext__(self):
    if not self._buf:
        page = await self._fetch(self.cursor)       # cancellation here: nothing changed yet
        self._buf.extend(page.items)                # update state only after the await
        self.cursor = page.next
    if not self._buf:
        raise StopAsyncIteration
    return self._buf.popleft()                      # no await between pop and return

The rule is to do every await before mutating state, so a cancellation leaves the iterator exactly as it was. That makes the iterator safe to use under asyncio.timeout() per item, which is exactly how timing out each item of an async iterator uses it. The generator-side equivalent is in making async iterators cancellation-safe.

Verify: wrap each anext() in a tiny timeout that sometimes fires; after retrying, the sequence of items received has no gaps and no duplicates.

Verification

The class iterator is correct when:

  • __aiter__ is synchronous and async for works without TypeError.
  • Exhaustion raises StopAsyncIteration, and anext(it, default) returns the default.
  • Cleanup is explicit — aclose() or async with — and happens when the consumer stops, not at GC.
  • A cancelled __anext__ leaves no item lost or duplicated.

Diagnostic Hook: expose iterator state as metrics when the iterator wraps an external stream — items buffered, pages fetched, current cursor age. A buffer that only grows means the consumer is slower than the producer; a cursor that stops advancing while the consumer is still iterating means the source has stalled and the iterator is waiting forever on a fetch without a timeout.

Pitfalls & edge cases

  • async def __aiter__. Breaks async for with a TypeError; it must be a plain method.
  • Raising StopIteration. Becomes RuntimeError inside a coroutine; use StopAsyncIteration.
  • Sharing one iterator between tasks. Two consumers calling __anext__ concurrently interleave fetches and can double-fetch a page; guard with a lock or give each consumer its own iterator.
  • Relying on __del__ for async cleanup. It cannot await; anything it tries to schedule may run after the loop has closed.

Frequently Asked Questions

How do I write an async iterator class in Python?

Implement aiter as a regular method that returns the iterator (often self), and anext as an async method that returns the next value or raises StopAsyncIteration when there are no more.

Should I use an async generator or a class with anext?

Use an async generator by default. Use a class when callers need visible state such as a cursor, operations like peek or push-back, or explicit control over cleanup. The per-item cost is effectively the same.

What do the aiter and anext builtins do?

aiter(obj) returns the async iterator for obj, like iter() for sync code. anext(it) returns an awaitable for the next item, and anext(it, default) returns the default instead of raising StopAsyncIteration when the iterator is exhausted. Both were added in Python 3.10.

Why does async for raise TypeError about aiter returning a coroutine?

Because aiter was declared with async def. It must be a plain method that returns an object with anext.