Cursor Pagination for Sensor History Endpoints #

Every sensor history endpoint is a moving target: new readings arrive while you are reading it. That single fact is what makes offset pagination wrong and cursor pagination right, and it is also why neither one can tell you whether you received everything. This guide covers walking a cursor feed correctly, persisting position so a restart resumes rather than restarts, and adding the completeness check that pagination itself cannot provide. It pairs with rate limit handling inside the REST API polling stage.

Why the Cursor Exists #

An offset is a position in an ordered result set — “skip the first 100 rows”. It is stable only if the set does not change between requests, which is the one thing a live sensor feed guarantees will not hold.

Consider a feed sorted newest-first, 50 readings per page. You fetch page one at 10:00:00 and receive readings 1–50. Between then and your request for page two, twelve new readings arrive. Now offset=50 skips what are currently the first 50 rows, which includes the twelve new ones plus readings 1–38 — so readings 39–50, which you already have, come back again, and nothing is lost. Sort oldest-first and the mirror image happens: rows shift the other way and you skip readings entirely. Either way, the page boundary is a position in a set that has moved.

A cursor is a position in the data, not in the result set: an opaque token encoding “after reading X”. New arrivals do not move X. Whatever inserts happen between requests, the next page starts exactly where the previous one ended. Every property that makes cursor paging safe follows from that one difference, and it is why the pagination comparison in the parent stage puts stability at the top of the table.

Production-Ready Implementation #

The walker is a generator so the caller processes pages as they arrive rather than accumulating a month of readings in memory — the same discipline as chunked I/O.

# python 3.11 · httpx==0.27.0
from __future__ import annotations

from collections.abc import Iterator
from dataclasses import dataclass

import httpx


@dataclass(frozen=True)
class Page:
    readings: list[dict]
    next_cursor: str | None
    newest_observed_at: str | None


def fetch_page(client: httpx.Client, url: str, cursor: str | None, page_size: int) -> Page:
    params = {"limit": page_size}
    if cursor:
        params["cursor"] = cursor
    r = client.get(url, params=params, timeout=30.0)
    r.raise_for_status()
    body = r.json()
    readings = body.get("results", [])
    return Page(
        readings=readings,
        next_cursor=body.get("next_cursor") or None,
        newest_observed_at=max((x["observed_at"] for x in readings), default=None),
    )


def walk(
    client: httpx.Client,
    url: str,
    start_cursor: str | None,
    *,
    page_size: int = 500,
    max_pages: int = 500,
) -> Iterator[Page]:
    """Yield pages until the feed is exhausted, guarding against a stuck cursor."""
    cursor, seen = start_cursor, set()
    for _ in range(max_pages):
        page = fetch_page(client, url, cursor, page_size)
        yield page
        if page.next_cursor is None:
            return
        if page.next_cursor in seen:
            raise RuntimeError(f"cursor loop detected at {page.next_cursor!r}")
        seen.add(page.next_cursor)
        cursor = page.next_cursor
    raise RuntimeError(f"exceeded {max_pages} pages — cursor is probably not advancing")

The loop detection is not paranoia. A server bug that returns the same next_cursor forever turns a polling job into an infinite loop that quietly re-inserts the same page until someone notices the duplicate counter climbing — and with an idempotent sink, nothing else will complain.

Persisting position is the other half. The cursor resumes; the timestamp recovers:

CHECKPOINT_SQL = """
INSERT INTO poll_checkpoint (feed, cursor, watermark, updated_at)
VALUES (%(feed)s, %(cursor)s, %(watermark)s, now())
ON CONFLICT (feed) DO UPDATE
SET cursor = EXCLUDED.cursor,
    watermark = GREATEST(poll_checkpoint.watermark, EXCLUDED.watermark),
    updated_at = now()
"""

def poll_once(conn, client: httpx.Client, feed: str, url: str) -> int:
    """One polling pass. Checkpoints after every page, not at the end."""
    with conn.cursor() as cur:
        cur.execute("SELECT cursor, watermark FROM poll_checkpoint WHERE feed = %s", (feed,))
        row = cur.fetchone()
    cursor, watermark = (row or (None, None))

    total = 0
    for page in walk(client, url, cursor):
        write_batch(conn, page.readings)          # idempotent on (device_id, observed_at, metric)
        total += len(page.readings)
        with conn.cursor() as cur:
            cur.execute(CHECKPOINT_SQL, {
                "feed": feed,
                "cursor": page.next_cursor,
                "watermark": page.newest_observed_at or watermark,
            })
        conn.commit()                              # checkpoint and rows commit together
    return total

Checkpointing per page inside the same transaction as the write is what makes a crash safe: you either have the page and its cursor, or neither. Checkpointing at the end of the whole walk means a crash on page 40 replays 39 pages — harmless with an idempotent sink, but slow and a waste of the rate-limit budget.

Parameter Tuning Guide #

The overlap in the recovery column is deliberate. Restarting slightly before the watermark re-requests readings you already have, and the idempotent sink absorbs them — that is much cheaper than the alternative failure, which is a silent gap at the seam.

Verification and Testing #

The completeness check is the test that matters, because it is the only one that catches a page you never knew existed.

# python 3.11 · pandas==2.2.2
import pandas as pd

def coverage_report(readings: pd.DataFrame, expected_per_hour: dict[str, int]) -> pd.DataFrame:
    """Readings received per sensor-hour against each sensor's expected cadence.

    Returns rows with completeness below 1.0 — candidate gaps, whatever their cause.
    """
    hourly = (
        readings.assign(hour=readings["observed_at"].dt.floor("h"))
        .groupby(["device_id", "hour"], as_index=False)
        .size()
        .rename(columns={"size": "received"})
    )
    hourly["expected"] = hourly["device_id"].map(expected_per_hour)
    hourly["completeness"] = hourly["received"] / hourly["expected"]
    return hourly[hourly["completeness"] < 1.0].sort_values("completeness")

Pair it with a walker test against a stub feed that inserts rows mid-walk — the exact condition offset paging fails on:

@respx.mock
def test_cursor_walk_is_stable_when_new_rows_arrive_mid_walk():
    pages = [
        {"results": [{"id": i} for i in range(1, 51)], "next_cursor": "c50"},
        # 12 new readings arrived here; a cursor feed still resumes after id 50
        {"results": [{"id": i} for i in range(51, 101)], "next_cursor": None},
    ]
    route = respx.get("https://api.example/history")
    route.side_effect = [httpx.Response(200, json=p) for p in pages]

    ids = [r["id"] for page in walk(httpx.Client(), "https://api.example/history", None)
           for r in page.readings]
    assert ids == list(range(1, 101))          # no skips, no repeats

Gotchas #

Treating an opaque cursor as parseable. Cursors are frequently base64-encoded JSON, and it is tempting to decode one to extract a timestamp. Vendors change the encoding without notice because it is documented as opaque. Persist it as a string and never inspect it.

Cursors that expire. Many APIs invalidate a cursor after minutes or hours. A poller that has been down over a weekend will get a 400 on resume. Catch that specific failure and fall back to the watermark path rather than crash-looping.

Assuming next_cursor: null means “you have everything”. It means “no more pages right now”. Readings that arrive a second later need another poll — which is exactly why the checkpoint records the cursor rather than a completion flag.

Sorting the feed newest-first for incremental polling. It works, but it makes the resume semantics harder to reason about and interacts badly with late-arriving data. Poll oldest-first from the cursor and let a separate pass handle the tail of late telemetry.

FAQ #

Related #

• REST API Polling and Batch IoT Ingestion — the polling stage this pagination belongs to
• Handling Rate Limits and Backoff When Polling Sensor APIs — keeping the request rate inside the API’s budget while you walk pages
• Automating QC Flags for Missing Environmental Readings — turning a detected coverage gap into a flag the dataset carries