Skip to content

Writing Async Iterators for Paginated APIs

Every list endpoint worth calling is paginated, and every codebase that calls several of them grows the same loop in a dozen places: fetch a page, process its items, read the next_cursor, stop when it is missing, remember to handle the empty page, forget to handle the rate limit. Worse, callers that only need the first matching record still download every page, because the loop was written to collect everything into a list first. An async iterator moves pagination into one place and gives callers the interface they actually want — async for invoice in list_invoices(customer): — with items arriving as soon as each page does, early break stopping further requests, and retries and timeouts applied per page. This guide builds that iterator as an async generator, makes early exit release its resources deterministically, adds one-page-ahead prefetching for throughput, and converts it to a resumable class for jobs that must survive restarts.

Prerequisites

Pagination hidden behind async for 4 stages from caller to next cursor. Pagination hidden behind async for caller async for item fetch page cursor in yield items one at a time next cursor or stop Callers see a stream of items; cursors never leave the iterator.

1. Yield items page by page with an async generator

Write the pagination loop once, inside an async def that yields individual items. Callers iterate with async for, and each page is requested only when the previous page's items have been consumed.

import asyncio
from dataclasses import dataclass
from typing import AsyncIterator

CALLS: list[str | None] = []


@dataclass
class Page:
    items: list[dict]
    next_cursor: str | None


async def fetch_page(customer: str, cursor: str | None, limit: int = 3) -> Page:
    """Fake cursor-paginated API: 10 invoices, pages of `limit`."""
    CALLS.append(cursor)
    await asyncio.sleep(0.01)                                # network round trip
    start = int(cursor or 0)
    items = [{"id": f"inv-{i}", "customer": customer} for i in range(start, min(start + limit, 10))]
    nxt = str(start + limit) if start + limit < 10 else None
    return Page(items, nxt)


async def list_invoices(customer: str) -> AsyncIterator[dict]:
    cursor: str | None = None
    while True:
        page = await fetch_page(customer, cursor)
        for item in page.items:
            yield item
        if page.next_cursor is None:                        # last page, even if it was empty
            return
        cursor = page.next_cursor


async def main() -> None:
    ids = [inv["id"] async for inv in list_invoices("acme")]
    print(len(ids), "invoices in", len(CALLS), "requests")  # 10 invoices in 4 requests


asyncio.run(main())

Deciding termination from next_cursor rather than from an empty page matters: some APIs return empty intermediate pages when results are filtered server-side, and some return a full last page with no cursor. The generator encodes that rule once.

Verify: the run prints 10 invoices in 4 requests, and CALLS shows the cursors [None, '3', '6', '9'] in order.

2. Stop requesting pages when the caller breaks early

The main payoff of a generator is laziness: a caller that finds what it needs can break, and no further pages are fetched. The subtle part is what happens to the generator afterwards. Breaking out of async for leaves it suspended at a yield; any cleanup in its finally — closing a response, releasing a connection — runs only when it is closed, which otherwise happens later during garbage collection or loop shutdown. Close it deterministically with contextlib.aclosing().

import asyncio
import contextlib


async def list_invoices_with_cleanup(customer: str):
    print("  opened listing session")
    try:
        async for item in list_invoices(customer):
            yield item
    finally:
        print("  closed listing session")                  # release connections, spans, etc.


async def find_first(customer: str, wanted: str) -> dict | None:
    async with contextlib.aclosing(list_invoices_with_cleanup(customer)) as invoices:
        async for inv in invoices:
            if inv["id"] == wanted:
                return inv                                   # stops paging here
    return None


async def main() -> None:
    CALLS.clear()
    print(await find_first("acme", "inv-4"))
    print("requests made:", len(CALLS))                      # 2: pages 0-2 and 3-5


asyncio.run(main())

closed listing session prints before the result, because aclosing closes the generator as the async with block exits. Only two requests were made for ten invoices. The detailed mechanics of generator finalisation are in closing async generators with aclosing.

Verify: the output shows the session opened, closed, then the invoice, and requests made: 2.

3. Prefetch the next page while items are consumed

A strictly lazy iterator is sequential: the request for page two does not start until the caller has processed every item of page one. When processing each item is itself slow, fetching the next page in the background, while the caller works, cuts total time. Keep the look-ahead to exactly one page so memory and upstream load stay bounded, and cancel the pending fetch if the caller stops early.

import asyncio
import contextlib


async def list_invoices_prefetch(customer: str):
    next_fetch = asyncio.create_task(fetch_page(customer, None))
    try:
        while True:
            page = await next_fetch
            if page.next_cursor is not None:
                next_fetch = asyncio.create_task(fetch_page(customer, page.next_cursor))  # one ahead
            for item in page.items:
                yield item
            if page.next_cursor is None:
                return
    finally:
        if not next_fetch.done():
            next_fetch.cancel()                              # caller stopped: drop the look-ahead
            await asyncio.gather(next_fetch, return_exceptions=True)


async def slow_process(item: dict) -> None:
    await asyncio.sleep(0.01)


async def consume(iterator_factory) -> float:
    loop = asyncio.get_running_loop()
    started = loop.time()
    async with contextlib.aclosing(iterator_factory("acme")) as items:
        async for item in items:
            await slow_process(item)
    return loop.time() - started


async def main() -> None:
    print(f"lazy:     {await consume(list_invoices):.3f}s")
    print(f"prefetch: {await consume(list_invoices_prefetch):.3f}s")


asyncio.run(main())

With a 10 ms page fetch and 10 ms per item, the lazy version takes roughly 140 ms and the prefetching version roughly 110 ms: the three later page fetches overlap with item processing instead of adding to it. The gain grows with page latency. It also changes failure timing — an error fetching page three can now surface while the caller is still on page two — which is usually what you want.

Verify: the prefetch timing is lower by about one page fetch per page boundary, and breaking out after the first item leaves no pending fetch_page task in asyncio.all_tasks().

Lazy fetching versus one-page look-ahead 4 lanes over time. Lazy fetching versus one-page look-ahead lazy: fetch p1 p2 p3 lazy: process items of p1 items of p2 items of p3 prefetch: fetch p1 p2 p3 prefetch: process items of p1 items of p2 items of p3 time → Each page boundary saves one fetch of waiting; memory holds at most one extra page.

4. Make the iterator resumable for long jobs

A nightly export that walks millions of records must survive a restart without starting over. A generator hides its cursor; a class implementing __aiter__ and __anext__ can expose it, so the job can checkpoint the cursor of the last fully processed page.

import asyncio


class InvoicePager:
    """Async iterator with an explicit, checkpointable cursor."""

    def __init__(self, customer: str, cursor: str | None = None) -> None:
        self.customer = customer
        self.page_cursor = cursor          # cursor that produced the current buffer
        self._next_cursor = cursor
        self._buffer: list[dict] = []
        self._exhausted = False

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

    async def __anext__(self) -> dict:
        while not self._buffer:
            if self._exhausted:
                raise StopAsyncIteration
            self.page_cursor = self._next_cursor
            page = await fetch_page(self.customer, self._next_cursor)
            self._buffer = list(page.items)
            self._next_cursor = page.next_cursor
            self._exhausted = page.next_cursor is None
        return self._buffer.pop(0)

    def checkpoint(self) -> str | None:
        """Cursor to resume from: replays the current page, never skips an item."""
        return self.page_cursor


async def main() -> None:
    pager = InvoicePager("acme")
    first_run = [await anext(pager) for _ in range(5)]      # crash after 5 items
    saved = pager.checkpoint()
    resumed = [inv["id"] async for inv in InvoicePager("acme", cursor=saved)]
    print(saved, resumed[:3])                                # '3' ['inv-3', 'inv-4', 'inv-5']


asyncio.run(main())

Checkpointing the cursor of the current page means a resume may reprocess a few items, but never skips any — so item processing must be idempotent, the same property idempotency keys give remote side effects.

Generator or class-based iterator? A grid of 4 rows by 2 columns. Generator or class-based iterator? need async generator class with __anext__ shortest code yes no cleanup via finally yes, with aclosing explicit aclose method expose a cursor awkward yes resume after restart awkward yes Start with a generator; switch to a class when the cursor must be visible.

Verify: after consuming five items, the checkpoint is '3', and resuming from it yields inv-3 onwards: items 3 and 4 are replayed, nothing is lost.

5. Retry and time-bound each page, not the whole walk

A walk over hundreds of pages should not fail because one request hit a transient error, and a timeout around the entire async for would cut off a healthy long walk. Put retries and the deadline around the single page fetch, inside the iterator.

import asyncio
import random


class TransientError(Exception):
    pass


async def fetch_page_with_retries(customer: str, cursor: str | None, *, attempts: int = 4,
                                  per_page_timeout: float = 2.0) -> Page:
    for attempt in range(1, attempts + 1):
        try:
            async with asyncio.timeout(per_page_timeout):
                return await flaky_fetch(customer, cursor)
        except (TransientError, TimeoutError):
            if attempt == attempts:
                raise
            await asyncio.sleep(random.uniform(0, 0.05 * 2 ** attempt))   # full jitter
    raise AssertionError("unreachable")


_failures = {"3": 2}                                         # page at cursor 3 fails twice


async def flaky_fetch(customer: str, cursor: str | None) -> Page:
    if _failures.get(cursor or "", 0) > 0:
        _failures[cursor] -= 1
        raise TransientError(f"503 on cursor {cursor}")
    return await fetch_page(customer, cursor)


async def list_invoices_resilient(customer: str):
    cursor = None
    while True:
        page = await fetch_page_with_retries(customer, cursor)
        for item in page.items:
            yield item
        if page.next_cursor is None:
            return
        cursor = page.next_cursor


async def main() -> None:
    print(len([i async for i in list_invoices_resilient("acme")]))   # 10


asyncio.run(main())

The iterator's callers see an uninterrupted stream; transient failures are absorbed where they happen, with backoff following exponential backoff with jitter in asyncio. A persistent failure still propagates out of async for, with the checkpoint from step 4 telling the job where to resume.

Verify: the run prints 10 although the second page failed twice, and making the failure permanent raises TransientError from the async for after four attempts.

Verification

The paginated iterator is correct when:

  • Callers never see pagination: they iterate items with async for, with no cursor handling of their own.
  • Early exit stops traffic: breaking after the first match makes no further requests, and cleanup runs before the caller continues.
  • Look-ahead is bounded: prefetching holds at most one extra page and cancels it when iteration stops.
  • Long walks are resumable: a checkpointed cursor restarts without skipping items.
  • Failures are local: retries and timeouts wrap single page fetches, and persistent errors surface from the loop.

Pitfalls & edge cases

  • Collecting into a list first. [x async for x in pages] defeats laziness and loads everything into memory; iterate and process instead.
  • Offset pagination on changing data. Offsets skip or duplicate items when rows are inserted during the walk. Prefer server-issued cursors or keyset pagination on a stable sort key.
  • Unbounded prefetching. Fetching all pages concurrently to "go faster" hammers the upstream and removes backpressure. Keep look-ahead to one page, or use a bounded queue, as compared in async generators vs queues for streaming pipelines.
  • Sharing one iterator between tasks. Two tasks calling __anext__ on the same generator concurrently raises RuntimeError: anext(): asynchronous generator is already running. Give each consumer its own iterator, or fan items out through a queue.
  • Rate limits inside the loop. Honour 429 and Retry-After inside the page fetch, not in each caller, so every consumer of the iterator is automatically polite.

Frequently Asked Questions

How do I iterate over a paginated API with async for?

Write an async generator that loops over pages: await the page request, yield each item, and follow the next cursor until the API stops returning one. Callers then write async for item in generator, receive items as each page arrives, and trigger the next request only after consuming the current page's items.

Does breaking out of an async for loop stop the async generator?

It stops iteration, so no more pages are requested, but the generator stays suspended and its finally block runs only when it is closed or garbage-collected. Wrap it in contextlib.aclosing so the generator is closed, and its cleanup runs, as soon as the async with block exits.

Should an async paginator prefetch the next page?

Prefetching one page ahead helps when processing items takes noticeable time, because the next request overlaps with processing. Limit look-ahead to a single page to bound memory and upstream load, and cancel the pending fetch in a finally block when the consumer stops early.

How do I resume a paginated export after a crash?

Use an iterator that exposes the cursor of the page it is currently serving, checkpoint that cursor after processing, and start a new iterator from it on restart. This may replay a few items of the interrupted page but never skips any, so make item processing idempotent.