Iterate export pages with bounded continuation cursors

README.md

@@ -47,6 +47,25 @@ with ProoflineClient() as client:
     # Continue with cursor=page.next_cursor when it is not None.
 ```
 
+`iter_export_pages` follows the continuation cursors and yields each page of one
+export. It stops at the last page and raises `ProoflineError` when the page bound
+is reached, a cursor repeats or the settled horizon changes between pages.
+
+```python
+from datetime import UTC, datetime
+from uuid import UUID
+
+with ProoflineClient() as client:
+    pages = client.iter_export_pages(
+        UUID(os.environ["PROOFLINE_ORGANIZATION_ID"]), "reviews",
+        from_time=datetime(2026, 10, 1, tzinfo=UTC),
+        to_time=datetime(2026, 10, 25, tzinfo=UTC),
+    )
+    for page in pages:
+        for row in page.rows:
+            print(row["review_id"], row["status"])
+```
+
 ## Coverage uploads
 
 Use a repository-restricted token with `coverage:write` and current repository write

proofline/__init__.py

@@ -32,7 +32,7 @@
 from ._transport import VERSION as __version__
 from ._transport import ProoflineError
 from .client import AsyncProoflineClient, ProoflineClient, ReviewFilters
-from .exports import ExportDataset, ExportFormat, ExportPage
+from .exports import ExportDataset, ExportFormat, ExportPage, ExportTime
 from .outcomes import AssessmentGap, CheckGap, ReviewOutcome
 
 __all__ = [
@@ -51,6 +51,7 @@
     "ExportDataset",
     "ExportFormat",
     "ExportPage",
+    "ExportTime",
     "FileReviewRequest",
     "FileReviewResponse",
     "PatchReviewRequest",

proofline/client.py

@@ -69,7 +69,14 @@
 from ._generated.models.stop_response import StopResponse
 from ._generated.types import UNSET, Response, Unset
 from ._transport import ProoflineError
-from .exports import ExportDataset, ExportFormat, ExportPage, decode_export
+from .exports import (
+    ExportDataset,
+    ExportFormat,
+    ExportPage,
+    ExportTime,
+    decode_export,
+    export_request,
+)
 from .fixes import AsyncFixesClient, FixesClient
 from .knowledge import AsyncKnowledgeClient, KnowledgeClient
 from .outcomes import ReviewOutcome, outcome
@@ -171,18 +178,14 @@ def _next_cursor(page: ReviewPage, seen: set[str]) -> str | None:
         return None
     cursor = page.next_cursor
     if not isinstance(cursor, str) or not cursor or cursor in seen:
-        raise ProoflineError(
-            0,
-            "urn:proofline:sdk:invalid-pagination",
-            "The review page has no new continuation cursor.",
-        )
+        raise _invalid_pagination()
     seen.add(cursor)
     return cursor
 
 
 def _pagination_limit() -> ProoflineError:
     return ProoflineError(
-        0, "urn:proofline:sdk:pagination-limit", "Review pagination reached the configured bound."
+        0, "urn:proofline:sdk:pagination-limit", "Pagination reached the configured bound."
     )
 
 
@@ -251,39 +254,57 @@ def _call(
         except Exception:
             raise _malformed() from None
 
+    def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: ExportTime,
+        to_time: ExportTime,
+        format: ExportFormat = "ndjson",
+        cursor: str | None = None,
+        group_by: str | None = None,
+        max_pages: int = 100,
+    ) -> Iterator[ExportPage]:
+        pager = _ExportPager(cursor, max_pages)
+        while True:
+            page = self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=pager.cursor,
+                group_by=group_by,
+            )
+            pager.accept(page)
+            yield page
+            if not pager.advance(page):
+                return
+
     def export_reviews(
         self,
         organization_id: UUID,
         dataset: ExportDataset,
         *,
-        from_time: str,
-        to_time: str,
+        from_time: ExportTime,
+        to_time: ExportTime,
         format: ExportFormat = "ndjson",
         cursor: str | None = None,
         group_by: str | None = None,
     ) -> ExportPage:
         """Read one export page. Repeat with next_cursor and the same filters."""
-        if dataset not in (
-            "reviews",
-            "finding_outcomes",
-            "finding_validations",
-            "spend",
-            "findings",
-        ):
-            raise ValueError("Unknown export dataset.")
-        if format not in ("csv", "ndjson"):
-            raise ValueError("Export format must be csv or ndjson.")
-        params = {"from": from_time, "to": to_time, "format": format}
-        if cursor is not None:
-            params["cursor"] = cursor
-        if group_by is not None:
-            params["group_by"] = group_by
+        request = export_request(
+            organization_id,
+            dataset,
+            from_time=from_time,
+            to_time=to_time,
+            format=format,
+            cursor=cursor,
+            group_by=group_by,
+        )
         try:
-            response = self._http.get(
-                f"/api/v1/organizations/{organization_id}/exports/{dataset}",
-                params=params,
-                headers={"Accept": "text/csv" if format == "csv" else "application/x-ndjson"},
-            )
+            response = self._http.get(request.path, params=request.params, headers=request.headers)
             return decode_export(response, format)
         except httpx.HTTPError as error:
             raise _network(error) from None
@@ -585,38 +606,58 @@ async def _call(
         except Exception:
             raise _malformed() from None
 
+    async def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: ExportTime,
+        to_time: ExportTime,
+        format: ExportFormat = "ndjson",
+        cursor: str | None = None,
+        group_by: str | None = None,
+        max_pages: int = 100,
+    ) -> AsyncIterator[ExportPage]:
+        pager = _ExportPager(cursor, max_pages)
+        while True:
+            page = await self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=pager.cursor,
+                group_by=group_by,
+            )
+            pager.accept(page)
+            yield page
+            if not pager.advance(page):
+                return
+
     async def export_reviews(
         self,
         organization_id: UUID,
         dataset: ExportDataset,
         *,
-        from_time: str,
-        to_time: str,
+        from_time: ExportTime,
+        to_time: ExportTime,
         format: ExportFormat = "ndjson",
         cursor: str | None = None,
         group_by: str | None = None,
     ) -> ExportPage:
         """Read one export page. Repeat with next_cursor and the same filters."""
-        if dataset not in (
-            "reviews",
-            "finding_outcomes",
-            "finding_validations",
-            "spend",
-            "findings",
-        ):
-            raise ValueError("Unknown export dataset.")
-        if format not in ("csv", "ndjson"):
-            raise ValueError("Export format must be csv or ndjson.")
-        params = {"from": from_time, "to": to_time, "format": format}
-        if cursor is not None:
-            params["cursor"] = cursor
-        if group_by is not None:
-            params["group_by"] = group_by
+        request = export_request(
+            organization_id,
+            dataset,
+            from_time=from_time,
+            to_time=to_time,
+            format=format,
+            cursor=cursor,
+            group_by=group_by,
+        )
         try:
             response = await self._http.get(
-                f"/api/v1/organizations/{organization_id}/exports/{dataset}",
-                params=params,
-                headers={"Accept": "text/csv" if format == "csv" else "application/x-ndjson"},
+                request.path, params=request.params, headers=request.headers
             )
             return decode_export(response, format)
         except httpx.HTTPError as error:
@@ -877,3 +918,43 @@ async def wait_for_review(
             return outcome(last, url, "TIMED_OUT") or ReviewOutcome("TIMED_OUT", url)
         finally:
             _transport.DEADLINE.reset(deadline)
+
+
+class _ExportPager:
+    def __init__(self, cursor: str | None, max_pages: int) -> None:
+        if not 1 <= max_pages <= 1_000:
+            raise ValueError("Export pagination requires max_pages 1..1000.")
+        self.cursor = cursor
+        self._seen = set() if cursor is None else {cursor}
+        self._remaining = max_pages
+        self._horizon: str | None = None
+
+    def accept(self, page: ExportPage) -> None:
+        if self._horizon is not None and page.settled_until != self._horizon:
+            raise _horizon_changed()
+        self._horizon = page.settled_until
+
+    def advance(self, page: ExportPage) -> bool:
+        self._remaining -= 1
+        cursor = page.next_cursor
+        if cursor is None:
+            return False
+        if not cursor or cursor in self._seen:
+            raise _invalid_pagination()
+        if not self._remaining:
+            raise _pagination_limit()
+        self._seen.add(cursor)
+        self.cursor = cursor
+        return True
+
+
+def _horizon_changed() -> ProoflineError:
+    return ProoflineError(
+        0, "urn:proofline:sdk:invalid-response", "The export settled horizon changed between pages."
+    )
+
+
+def _invalid_pagination() -> ProoflineError:
+    return ProoflineError(
+        0, "urn:proofline:sdk:invalid-pagination", "The page has no new continuation cursor."
+    )

proofline/exports.py

@@ -1,20 +1,25 @@
-"""Bounded export pages, with their original text and continuation metadata."""
+"""Export requests and bounded export pages with their text and continuation metadata."""
 
 from __future__ import annotations
 
 import csv
 import io
 import json
+from collections.abc import Mapping
 from dataclasses import dataclass
-from datetime import datetime
-from typing import Any, Literal
+from datetime import UTC, datetime
+from typing import Any, Literal, get_args
+from uuid import UUID
 
 import httpx
 
 from ._transport import ProoflineError
 
 ExportDataset = Literal["reviews", "finding_outcomes", "finding_validations", "spend", "findings"]
 ExportFormat = Literal["csv", "ndjson"]
+ExportTime = str | datetime
+_DATASETS: frozenset[str] = frozenset(get_args(ExportDataset))
+_MEDIA_TYPES: Mapping[ExportFormat, str] = {"csv": "text/csv", "ndjson": "application/x-ndjson"}
 
 
 @dataclass(frozen=True)
@@ -27,11 +32,53 @@ class ExportPage:
     next_cursor: str | None
 
 
+@dataclass(frozen=True)
+class ExportRequest:
+    path: str
+    params: dict[str, str]
+    headers: dict[str, str]
+
+
+def export_request(
+    organization_id: UUID,
+    dataset: ExportDataset,
+    *,
+    from_time: ExportTime,
+    to_time: ExportTime,
+    format: ExportFormat,
+    cursor: str | None,
+    group_by: str | None,
+) -> ExportRequest:
+    if dataset not in _DATASETS:
+        raise ValueError("Unknown export dataset.")
+    if format not in _MEDIA_TYPES:
+        raise ValueError("Export format must be csv or ndjson.")
+    if cursor == "":
+        raise ValueError("The export cursor must be nonempty.")
+    params = {"from": export_time(from_time), "to": export_time(to_time), "format": format}
+    if cursor is not None:
+        params["cursor"] = cursor
+    if group_by is not None:
+        params["group_by"] = group_by
+    return ExportRequest(
+        f"/api/v1/organizations/{organization_id}/exports/{dataset}",
+        params,
+        {"Accept": _MEDIA_TYPES[format]},
+    )
+
+
+def export_time(value: ExportTime) -> str:
+    if isinstance(value, str):
+        return value
+    if value.utcoffset() is None:
+        raise ValueError("Export window datetimes must be timezone-aware.")
+    return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
+
+
 def decode_export(response: httpx.Response, format: ExportFormat) -> ExportPage:
     try:
-        media = "text/csv" if format == "csv" else "application/x-ndjson"
         if (
-            response.headers.get("content-type", "").split(";")[0].strip() != media
+            response.headers.get("content-type", "").split(";")[0].strip() != _MEDIA_TYPES[format]
             or response.headers.get("x-proofline-export-version") != "1"
         ):
             raise ValueError

scripts/contracts.py

@@ -29,7 +29,7 @@ def main() -> None:
             "The generated JSON operations must match the twenty-one public JSON token routes."
         )
     tree = ast.parse((ROOT / "proofline/client.py").read_text())
-    helpers = {"close", "aclose", "iter_reviews", "wait_for_review"}
+    helpers = {"close", "aclose", "iter_reviews", "iter_export_pages", "wait_for_review"}
     clients = {
         node.name: {
             member.name