Paginating Large SaaS Exports for DSR

Within the broader SaaS API Sync Strategies layer of the Cross-System Data Discovery & Sync architecture, one failure mode quietly ruins more Data Subject Request (DSR) access responses than rate limits do: a large multi-page export that skips or double-counts records because the data changed while it was being read, or that dies on page 400 of 900 and has to start over from scratch. The engineer who hits this is anyone extracting a high-volume subject — a long-tenured customer with tens of thousands of events across a ticketing, marketing, or billing SaaS — where a single naive offset/limit loop silently drops rows whenever a concurrent write shifts the window. This page covers the specific mechanics of paginating a large export safely: why cursor (keyset) pagination beats offset pagination over mutating data, how a resumable checkpoint lets an interrupted export continue rather than restart, how to deduplicate records that appear on more than one page, and how to stream rows so a big export does not blow up worker memory — all inside the response window GDPR Article 12(3) fixes at one month and CCPA §1798.130(a)(2) fixes at 45 days. It builds on the same throttle-aware discipline described in Handling Rate Limits in Salesforce API Sync; here the concern is not the vendor’s 429 but the integrity and resumability of the page loop itself.

The core idea is a loop that never holds more than one page in memory, records exactly where it is after every page, and can be killed and restarted at any point without losing or duplicating a record.

Prerequisites #

This implementation targets Python 3.11+ and the following libraries:

• httpx (0.27+) — an async HTTP client for the SaaS REST endpoints; the async generator below streams pages without buffering the whole export.
• pydantic (v2) — a frozen Checkpoint model serialized between pages so a resumed run validates its saved position before continuing. Uses v2 idioms (model_config = ConfigDict(...), model_validate, model_dump); v1 patterns will not work.
• A durable checkpoint store — Redis, a database row, or an object-store key. The in-code store below is an interface; production must back it with persistent storage so a crashed worker’s position survives.
• A stable, monotonic sort key on the source — a cursor is only safe if the API can page over an immutable ordering (a created timestamp plus a tie-breaking id). Without one, no pagination scheme is correct over mutating data.

Credentials are resolved at call time from a secret manager, and the export is a read-only, idempotent pass, consistent with the least-privilege posture the SaaS API Sync Strategies layer mandates.

Step-by-step implementation #

Step 1 — Model a resumable checkpoint with Pydantic v2 #

The checkpoint is the single fact that makes a large export resumable: it is the position the loop can restart from. Modeling it as a frozen Pydantic v2 record means a resumed run validates the saved state before trusting it, rather than paging blindly from a corrupt token.

from datetime import datetime, timezone
from pydantic import BaseModel, ConfigDict, Field


class Checkpoint(BaseModel):
    """A durable, resumable position in a large paginated export."""

    model_config = ConfigDict(frozen=True)

    job_id: str
    vendor: str
    cursor: str | None = None                 # opaque continuation token, None at start
    pages_done: int = 0
    records_seen: int = 0
    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

    def advance(self, next_cursor: str | None, page_records: int) -> "Checkpoint":
        """Return the next checkpoint after a page is durably streamed."""
        return self.model_copy(update={
            "cursor": next_cursor,
            "pages_done": self.pages_done + 1,
            "records_seen": self.records_seen + page_records,
            "updated_at": datetime.now(timezone.utc),
        })

Because the model is frozen, advance returns a new checkpoint rather than mutating the current one — so a checkpoint is only ever replaced after its page has been safely handed to the sink, never half-updated mid-page.

Step 2 — Stream pages through a generic cursor paginator #

The paginator is an async generator: it fetches exactly one page, yields it, and holds nothing else in memory. Cursor pagination is what keeps this correct over live data. An offset/limit loop re-counts from the top of a shifting result set on every request, so a row inserted or deleted earlier in the ordering slides every later row by one position — skipping a record or serving it twice. A cursor (keyset) anchors on the last row’s stable sort key, so inserts and deletes elsewhere cannot move the boundary the next page reads from.

from collections.abc import AsyncIterator
from typing import Any

import httpx


async def paginate(
    client: httpx.AsyncClient, path: str, start_cursor: str | None, page_size: int = 200
) -> AsyncIterator[tuple[list[dict[str, Any]], str | None]]:
    """Yield (rows, next_cursor) one page at a time; never buffer the whole export."""
    cursor = start_cursor
    while True:
        params: dict[str, Any] = {"limit": page_size}
        if cursor is not None:
            params["cursor"] = cursor          # opaque token, NOT a numeric offset
        response = await client.get(path, params=params)
        response.raise_for_status()
        body = response.json()

        rows = body.get("data", [])
        next_cursor = (body.get("paging") or {}).get("next_cursor")
        yield rows, next_cursor

        if not next_cursor:                    # exhausted -> loop ends
            return
        cursor = next_cursor

Yielding (rows, next_cursor) rather than returning a full list is the streaming guarantee: the caller processes and discards each page before the next request is made, so a subject with a million events costs one page of memory, not a million rows.

Step 3 — Persist the checkpoint between pages so the export survives interruption #

Streaming and checkpointing combine into the resumable loop. After each page is deduplicated and handed to the sink, the loop persists the next cursor. If the worker is killed — a deploy, an OOM, a spot-instance reclaim — the next run loads the checkpoint and continues from the saved cursor instead of restarting the whole export and burning SLA budget re-reading pages it already has.

from typing import Protocol

import httpx


class CheckpointStore(Protocol):
    """Durable persistence for a resumable export position."""
    async def load(self, job_id: str) -> Checkpoint | None: ...
    async def save(self, checkpoint: Checkpoint) -> None: ...


async def run_export(
    client: httpx.AsyncClient,
    store: CheckpointStore,
    sink: "RecordSink",
    job_id: str,
    vendor: str,
    path: str,
) -> Checkpoint:
    """Resume from the last checkpoint (if any) and export to completion."""
    checkpoint = await store.load(job_id) or Checkpoint(job_id=job_id, vendor=vendor)
    dedup = DedupFilter()

    async for rows, next_cursor in paginate(client, path, checkpoint.cursor):
        fresh = [r for r in rows if dedup.first_time(r)]
        await sink.write(fresh)                 # stream this page to durable storage
        checkpoint = checkpoint.advance(next_cursor, len(fresh))
        await store.save(checkpoint)            # persist AFTER the page is durable

    return checkpoint

The ordering is deliberate and load-bearing: the checkpoint is saved after the sink write, so the persisted position never claims a page the sink has not yet stored. On resume, at worst the last page is re-fetched and its records are dropped by the dedup filter — an idempotent replay, never a gap. This is the same “record job state on a durable store” discipline the Async Polling & Queue Management layer applies to the whole discovery lifecycle.

Step 4 — Deduplicate by a stable record identifier #

Even with cursor pagination, a page boundary that lands on rows sharing the same sort-key timestamp, or a resumed run that replays its last page, will surface the same record twice. Deduplication keys on a stable identifier — the vendor’s immutable record id — not on page position or content hash, so the manifest counts distinct records and an access response under GDPR Article 15 does not list the same event repeatedly.

class DedupFilter:
    """Suppresses records already seen in this export by stable id."""

    def __init__(self) -> None:
        self._seen: set[str] = set()

    def first_time(self, record: dict) -> bool:
        """True the first time a record id is seen; False on any repeat."""
        record_id = record.get("id")
        if record_id is None:
            raise ValueError("record missing stable 'id'; cannot deduplicate safely")
        if record_id in self._seen:
            return False
        self._seen.add(record_id)
        return True

For exports too large to hold every id in memory, back _seen with a persisted set (a Redis set keyed by job_id, or a UNIQUE constraint on the sink) so dedup state itself survives a resume. Raising on a missing id is deliberate: silently keeping an unkeyable record risks double-counting, and silently dropping it risks under-reporting — both are worse than a loud failure that routes the row to triage.

Configuration reference #

Verification #

Prove the loop is resumable and idempotent before trusting it with a real subject. The two critical assertions are that a run resumed from a mid-export checkpoint continues from the saved cursor, and that a replayed final page produces no duplicate records.

import pytest


@pytest.mark.asyncio
async def test_resume_continues_from_saved_cursor(fake_client, mem_store, mem_sink):
    """A checkpoint at page 2 makes the next run start at the saved cursor, not page 0."""
    await mem_store.save(Checkpoint(job_id="j1", vendor="demo", cursor="c2", pages_done=2))
    await run_export(fake_client, mem_store, mem_sink, "j1", "demo", "/events")
    assert fake_client.first_requested_cursor == "c2"   # never re-read pages 0-1


@pytest.mark.asyncio
async def test_replayed_page_is_deduplicated(fake_client_with_overlap, mem_store, mem_sink):
    """A resumed run that re-fetches its last page emits each record once."""
    await run_export(fake_client_with_overlap, mem_store, mem_sink, "j2", "demo", "/events")
    ids = [r["id"] for r in mem_sink.written]
    assert len(ids) == len(set(ids))                    # no duplicates in the manifest

In production, expect one structured log line per page carrying the job_id, pages_done, records_seen, and the cursor age, plus a final reconciliation line comparing records_seen against the vendor’s reported total. A mismatch is the signal that a page was silently skipped and the export must not be marked complete.

Troubleshooting #

Cursor expiry — a resumed run gets 400/410 on the saved cursor : Root cause: the vendor invalidated the continuation token because too long elapsed between pages, or the underlying snapshot rolled over. Fix: treat expiry as a resume-from-anchor signal — restart the affected object from the last stable sort key recorded in the checkpoint rather than from page zero, and shorten page_size so pages complete before the cursor’s TTL. Never silently stop, which would under-report the subject.

Duplicate rows in the manifest : Root cause: dedup keyed on page position or a content hash instead of the stable id, so a replayed final page or a tie on the sort key slips a record through twice. Fix: key DedupFilter on the vendor’s immutable record id and, for exports larger than memory, persist the seen-set so it survives a resume.

Records silently missing after a run marked complete : Root cause: offset pagination over live data — a concurrent insert shifted the window and a row fell between pages. Fix: switch the loop to cursor (keyset) pagination anchored on a monotonic key, and reconcile records_seen against the vendor total before marking the manifest complete; a mismatch must block completion.

Unstable ordering skips or repeats rows : Root cause: the API sorts on a non-unique or mutable field (a bare timestamp, or updated_at), so two rows can swap order between pages. Fix: page over an immutable composite key — created-at plus a tie-breaking id — and never sort on a field the source can rewrite mid-export.

Worker memory grows with export size : Root cause: the pages were collected into a list (or the async generator was drained into one) instead of streamed. Fix: consume paginate one page at a time and write each page to the sink before requesting the next, so peak memory is one page regardless of subject volume.

Frequently Asked Questions #

Related #

• SaaS API Sync Strategies — the parent guide to credential isolation, cursor normalization, and SLA telemetry across heterogeneous SaaS APIs.
• Handling Rate Limits in Salesforce API Sync — the throttle-aware backoff and dead-letter discipline that wraps this page loop.
• Async Polling & Queue Management — the durable job-state lifecycle these checkpoints participate in.
• Schema Validation Rules — the gate that validates each streamed, deduplicated page before it enters the manifest.
• Cross-System Data Discovery & Sync — the parent architecture this export loop feeds.