Paginating API Results in a Python CLI

mytool builds --status failed prints nothing, so the user concludes the pipeline is healthy. In fact the API returns 50 builds per page, the command only ever asked for the first page, and all 12 failures were on page three. Pagination bugs are the quietest bugs an API-backed CLI can have: nothing errors, the output looks plausible, and the answer is wrong. The opposite mistake is nearly as bad — a command that dutifully fetches all 40,000 records into a list before printing the first row, taking a minute and a gigabyte to show the user the ten items they wanted. This guide shows how to wrap any pagination scheme in a generator, expose --limit and --all sensibly, stream results to the terminal or to scripts, and test it all offline. It is part of the HTTP APIs topic.

Prerequisites

• Python 3.10+, httpx and Typer.
• An API client built around one httpx.Client, as in building an API client CLI with httpx.
• The API's pagination documentation — you need to know which of the three styles below it uses.

Three ways APIs paginate

Almost every REST API uses one of three schemes. They differ in how you ask for the next page and how you know you have reached the end:

• Page numbers or offsets (?page=3&per_page=50, ?offset=100&limit=50). Easy to understand; fragile when data changes between requests, because inserting an item shifts everything and can make you skip or repeat one.
• Cursors (?cursor=eyJpZCI6...). The response includes an opaque token pointing at the next page. Stable under concurrent changes and efficient for the server; you cannot jump to page 40, which a CLI rarely needs.
• Link headers (Link: <https://api/...&page=4>; rel="next"), used by GitHub and others. The server hands you the full URL of the next page; httpx parses the header into response.links.

The recipe: generators that yield items

Whatever the scheme, wrap it in a generator that yields individual items, not pages. The caller then never thinks about pages at all — it iterates, and fetching happens lazily as it goes.

# src/mytool/paging.py
from __future__ import annotations

from collections.abc import Iterator
from typing import Any

import httpx

Item = dict[str, Any]


def iter_cursor(client: httpx.Client, path: str, params: dict[str, Any] | None = None,
                *, page_size: int = 100) -> Iterator[Item]:
    """Cursor pagination: {"items": [...], "next_cursor": "..." | null}."""
    query = {**(params or {}), "limit": page_size}
    while True:
        response = client.get(path, params=query)
        response.raise_for_status()
        body = response.json()
        yield from body["items"]
        cursor = body.get("next_cursor")
        if not cursor:
            return
        query["cursor"] = cursor


def iter_pages(client: httpx.Client, path: str, params: dict[str, Any] | None = None,
               *, page_size: int = 100) -> Iterator[Item]:
    """Page-number pagination: a short page means the end."""
    page = 1
    while True:
        response = client.get(path, params={**(params or {}), "page": page, "per_page": page_size})
        response.raise_for_status()
        items = response.json()
        yield from items
        if len(items) < page_size:
            return
        page += 1


def iter_links(client: httpx.Client, path: str, params: dict[str, Any] | None = None) -> Iterator[Item]:
    """RFC 8288 Link headers: follow rel="next" until it disappears."""
    response = client.get(path, params=params)
    while True:
        response.raise_for_status()
        yield from response.json()
        next_link = response.links.get("next", {}).get("url")
        if not next_link:
            return
        response = client.get(next_link)

Two details are easy to miss. In iter_pages, the end is detected by a short page, which costs no extra request when the last page is partial — but if the total is an exact multiple of the page size, you make one final request that returns an empty list, which is correct and cheap. And in iter_links, the next URL is absolute and already contains the query string, so you pass it as-is rather than re-adding parameters.

Limits, filtering and "all"

Now the command decides how much to consume. itertools.islice stops the generator after limit items, and because the generator is lazy, stopping early means not requesting the remaining pages:

# src/mytool/cli.py
import itertools
import json
import os

import httpx
import typer

from mytool.paging import iter_cursor

app = typer.Typer()


@app.callback()
def main(ctx: typer.Context) -> None:
    """Build history."""
    client = httpx.Client(base_url=os.environ.get("MYTOOL_API_URL", "https://api.example.com/v1"),
                          timeout=httpx.Timeout(30.0, connect=5.0))
    ctx.call_on_close(client.close)
    ctx.obj = client


@app.command()
def builds(
    ctx: typer.Context,
    status: str | None = typer.Option(None, help="Only builds with this status."),
    limit: int = typer.Option(20, "--limit", "-n", min=1, help="Show at most N builds."),
    all_: bool = typer.Option(False, "--all", help="Fetch every matching build."),
    json_lines: bool = typer.Option(False, "--json-lines", help="One JSON object per line."),
) -> None:
    """List recent builds, newest first."""
    params = {"status": status} if status else {}
    items = iter_cursor(ctx.obj, "/builds", params)
    if not all_:
        items = itertools.islice(items, limit)
    shown = 0
    for b in items:
        shown += 1
        if json_lines:
            typer.echo(json.dumps(b, separators=(",", ":")))
        else:
            typer.echo(f"#{b['id']:<6} {b['project']:<10} {b['status']:<8} {b['age']}")
    if not json_lines and not all_ and shown == limit:
        typer.echo(f"(showing {limit}; use --limit or --all for more)", err=True)


if __name__ == "__main__":
    app()

Filtering belongs on the server whenever the API supports it — pass status=failed as a query parameter rather than fetching everything and filtering in Python. Server-side filtering is what makes "show me the last 20 failures" one request instead of twenty.

UX considerations

• Default to a small limit for humans. Twenty or fifty rows fit on a screen and return in one request. Tell the user when output was truncated and how to get more, as the final stderr line above does.
• Make "everything" explicit. --all is a deliberate choice that might take a while; it should never be the default for an unbounded collection.
• Stream, do not collect. Print each row as it arrives. The user sees results immediately, and memory stays flat no matter how many pages there are.
• Offer a streaming machine format. A JSON array cannot be written until the last page arrives. NDJSON — one JSON object per line — can be written incrementally and consumed incrementally by jq, grep or another program, which is why --json-lines suits paginated commands. The broader contract is in emitting JSON output for scripting.
• Show progress for long walks. For --all over thousands of items, a count on stderr ("fetched 3,200 builds...") updated in place tells the user the tool is working. Keep it off when stderr is not a terminal.
• Survive a downstream head. mytool builds --all --json-lines | head -5 closes the pipe after five lines; handle BrokenPipeError so the command exits quietly and stops fetching. See handling broken pipe and SIGPIPE.

Testing the behaviour

A fake API that serves a known dataset in pages lets you assert both the items returned and the number of requests made — the latter is what proves --limit stops early:

# tests/test_paging.py
import itertools

import httpx

from mytool.paging import iter_cursor, iter_links, iter_pages

DATA = [{"id": i} for i in range(1, 251)]   # 250 items


def cursor_api(calls):
    def handler(request):
        calls.append(request)
        start = int(request.url.params.get("cursor", 0))
        size = int(request.url.params["limit"])
        chunk = DATA[start:start + size]
        nxt = start + size if start + size < len(DATA) else None
        return httpx.Response(200, json={"items": chunk, "next_cursor": nxt and str(nxt)})
    return httpx.Client(base_url="https://api.test", transport=httpx.MockTransport(handler))


def test_cursor_walks_everything():
    calls = []
    items = list(iter_cursor(cursor_api(calls), "/builds", page_size=100))
    assert [i["id"] for i in items] == list(range(1, 251))
    assert len(calls) == 3


def test_limit_stops_fetching():
    calls = []
    first = list(itertools.islice(iter_cursor(cursor_api(calls), "/builds", page_size=100), 20))
    assert len(first) == 20
    assert len(calls) == 1


def test_page_numbers_exact_multiple():
    def handler(request):
        page = int(request.url.params["page"])
        return httpx.Response(200, json=DATA[:200][(page - 1) * 100: page * 100])
    client = httpx.Client(base_url="https://api.test", transport=httpx.MockTransport(handler))
    assert len(list(iter_pages(client, "/b", page_size=100))) == 200


def test_link_headers():
    def handler(request):
        page = int(request.url.params.get("page", 1))
        headers = {"Link": f'<https://api.test/b?page={page + 1}>; rel="next"'} if page < 3 else {}
        return httpx.Response(200, json=[{"page": page}], headers=headers)
    client = httpx.Client(base_url="https://api.test", transport=httpx.MockTransport(handler))
    assert [i["page"] for i in iter_links(client, "/b")] == [1, 2, 3]

The "exact multiple" test pins the edge case where the final page is full and the generator must make one more request to discover the end. For more on offline HTTP tests, see mocking filesystem and network in CLI tests.

Conclusion

Hide pagination behind generators that yield items, and let commands decide how many to take. Small default limits with a clear "showing N" note, an explicit --all, server-side filtering, streaming output and NDJSON for scripts together make listing commands both correct and fast. The generator shape also composes with everything else: retries per page, progress counters, and early exit when a pipe closes.

Frequently asked questions

Related

• Up: Calling HTTP APIs from Python CLIs
• Building an API client CLI with httpx
• Retries and backoff for CLI HTTP calls
• Processing large files and NDJSON streams
• Emitting JSON output for scripting