lesson 153 HTTP & API Clients · advanced

Pagination: follow the pages to the end

APIs return big results a page at a time. The client asks for the first page, reads how to ask for the next one (a page number, an offset or a cursor), and stops when the API says there is nothing more.

Simulated: this runs against results recorded on real Python.

You'll be able to

Recognise page-number, offset and cursor pagination
Loop until the end signal: an empty or short page, or no next cursor
Guard the loop with a page limit so a bug or a changing result can't run forever

The idea

OpenAlex has 53,612 works mentioning Python from 2019–2024. No API sends that in one response; it sends pages. Three styles are common:

page numbers: page=1, page=2, … with per-page results each, until a page comes back short or empty;
offset and limit: offset=1&limit=100, then offset=101, … (the USGS counts offsets from 1);
cursors: the first request sends cursor=*, and each response's meta.next_cursor is the token for the next page, None at the end. A cursor stays correct even when new records arrive between requests, which is why big APIs prefer it.

All three have the same loop: request a page, keep its results, work out the next request, stop on the end signal.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Page numbers until a short page

import io
import json
from email.message import Message
from http import HTTPStatus
from urllib.request import BaseHandler, Request, build_opener
from urllib.response import addinfourl

class Replay(BaseHandler):
    """Answers https requests from saved copies of real API responses."""
    handler_order = 100

    def __init__(self, saved):
        self.saved = saved

    def https_open(self, request):
        status, body = self.saved.get(request.full_url, (404, {"error": "Not Found"}))
        headers = Message()
        headers["Content-Type"] = "application/json"
        response = addinfourl(io.BytesIO(json.dumps(body).encode()), headers, request.full_url, status)
        response.msg = HTTPStatus(status).phrase
        return response
from urllib.parse import urlencode

base = "https://api.openalex.org/works?"
def page_url(page):
    return base + urlencode({"search": "python", "per-page": 2, "page": page})

saved = {
    page_url(1): (200, {"meta": {"count": 5, "page": 1}, "results": [{"title": "SciPy 1.0"}, {"title": "PyTorch"}]}),
    page_url(2): (200, {"meta": {"count": 5, "page": 2}, "results": [{"title": "Array programming with NumPy"}, {"title": "Denoising Diffusion Probabilistic Models"}]}),
    page_url(3): (200, {"meta": {"count": 5, "page": 3}, "results": [{"title": "PsychoPy2"}]}),
}
opener = build_opener(Replay(saved))

titles = []
page = 1
while True:
    with opener.open(page_url(page), timeout=10) as response:
        results = json.loads(response.read())["results"]
    titles.extend([work["title"] for work in results])
    if len(results) < 2:
        break
    page += 1
print(page, "pages,", len(titles), "works")
print(titles[-1])

With a cursor, the response itself says what to ask for next. The loop keeps the cursor in a variable, and a max_pages limit protects against a bug that never updates it, or an API that keeps returning a next cursor.

example-2.py · Cursor pagination with a safety limit

import io
import json
from email.message import Message
from http import HTTPStatus
from urllib.request import BaseHandler, Request, build_opener
from urllib.response import addinfourl

class Replay(BaseHandler):
    """Answers https requests from saved copies of real API responses."""
    handler_order = 100

    def __init__(self, saved):
        self.saved = saved

    def https_open(self, request):
        status, body = self.saved.get(request.full_url, (404, {"error": "Not Found"}))
        headers = Message()
        headers["Content-Type"] = "application/json"
        response = addinfourl(io.BytesIO(json.dumps(body).encode()), headers, request.full_url, status)
        response.msg = HTTPStatus(status).phrase
        return response
from urllib.parse import urlencode

def all_results(opener, base_url, max_pages=100):
    results = []
    cursor = "*"
    for _ in range(max_pages):
        url = base_url + "&" + urlencode({"cursor": cursor})
        with opener.open(url, timeout=10) as response:
            page = json.loads(response.read())
        results.extend(page["results"])
        cursor = page["meta"]["next_cursor"]
        if cursor is None:
            return results
    raise RuntimeError(f"still more pages after {max_pages}")

base = "https://api.openalex.org/works?filter=publication_year:2021"
saved = {
    base + "&cursor=%2A": (200, {"meta": {"next_cursor": "IlsxNjk"}, "results": [{"id": "W3201073812"}, {"id": "W3038025793"}]}),
    base + "&cursor=IlsxNjk": (200, {"meta": {"next_cursor": "Ils3NTc"}, "results": [{"id": "W3168436232"}]}),
    base + "&cursor=Ils3NTc": (200, {"meta": {"next_cursor": None}, "results": []}),
}
works = all_results(build_opener(Replay(saved)), base)
print([w["id"] for w in works])

When to use it: whenever a response has next, cursor, page or offset in it, or its count is larger than the results you got. Mistakes to avoid: forgetting to advance the page or cursor (the same page forever); stopping one page early by testing the count before keeping the page's results; and assuming page numbers start at 0 or offsets at 0 without reading the documentation.

Review

Paginated APIs send results a page at a time: page numbers, offsets (USGS starts at 1) or cursors.
Keep each page's results, then stop on the end signal: a short or empty page, or no next cursor.
Limit the number of pages, and deduplicate by id when results can shift between requests.
Next: requests that are slow or fail for a moment, and how to retry them safely.

Practise this