Add lazy paginated export iterators for sync and async clients

README.md

@@ -47,6 +47,29 @@ with ProoflineClient() as client:
     # Continue with cursor=page.next_cursor when it is not None.
 ```
 
+Use `iter_export_pages` to follow continuation cursors lazily while keeping the
+same organization, dataset, window, format and grouping on every request:
+
+```python
+with ProoflineClient() as client:
+    for page in client.iter_export_pages(
+        UUID(os.environ["PROOFLINE_ORGANIZATION_ID"]), "reviews",
+        from_time="2026-10-01T00:00:00Z", to_time="2026-10-25T00:00:00Z",
+    ):
+        for row in page.rows:
+            print(row)
+```
+
+Use `async for` with `AsyncProoflineClient`. Each yielded page retains its raw
+text, decoded rows and metadata; CSV pages retain their individual headers.
+Pass `cursor=` to resume an export. Iteration stops only when the server omits
+the next cursor, including when an intermediate page has no rows. Empty or
+repeated continuation cursors raise `ProoflineError` with type
+`urn:proofline:sdk:invalid-pagination`. The default bound is 100 pages; set
+`max_pages` to an integer from 1 to 1000. Exhausting that bound while a cursor
+remains raises `urn:proofline:sdk:pagination-limit`, so a partial export cannot
+silently appear complete. Stop iteration early when only a prefix is needed.
+
 ## Coverage uploads
 
 Use a repository-restricted token with `coverage:write` and current repository write

proofline/client.py

@@ -180,12 +180,33 @@ def _next_cursor(page: ReviewPage, seen: set[str]) -> str | None:
     return cursor
 
 
-def _pagination_limit() -> ProoflineError:
+def _pagination_limit(resource: str = "Review") -> ProoflineError:
     return ProoflineError(
-        0, "urn:proofline:sdk:pagination-limit", "Review pagination reached the configured bound."
+        0,
+        "urn:proofline:sdk:pagination-limit",
+        f"{resource} pagination reached the configured bound.",
     )
 
 
+def _export_page_limit(max_pages: int) -> None:
+    if isinstance(max_pages, bool) or not isinstance(max_pages, int) or not 1 <= max_pages <= 1_000:
+        raise ValueError("Export pagination requires max_pages 1..1000.")
+
+
+def _export_next_cursor(page: ExportPage, seen: set[str]) -> str | None:
+    cursor = page.next_cursor
+    if cursor is None:
+        return None
+    if not cursor or cursor in seen:
+        raise ProoflineError(
+            0,
+            "urn:proofline:sdk:invalid-pagination",
+            "The export page has no new continuation cursor.",
+        )
+    seen.add(cursor)
+    return cursor
+
+
 class ProoflineClient:
     """Synchronous typed HTTP operations.
 
@@ -288,6 +309,37 @@ def export_reviews(
         except httpx.HTTPError as error:
             raise _network(error) from None
 
+    def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: str,
+        to_time: str,
+        format: ExportFormat = "ndjson",
+        cursor: str | None = None,
+        group_by: str | None = None,
+        max_pages: int = 100,
+    ) -> Iterator[ExportPage]:
+        """Stream export pages, refusing cursor cycles and incomplete bounded traversals."""
+        _export_page_limit(max_pages)
+        seen = {cursor} if cursor is not None else set()
+        for _ in range(max_pages):
+            page = self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=cursor,
+                group_by=group_by,
+            )
+            cursor = _export_next_cursor(page, seen)
+            yield page
+            if cursor is None:
+                return
+        raise _pagination_limit("Export")
+
     def list_reviews(self, organization_id: UUID, **filters: Unpack[ReviewFilters]) -> ReviewPage:
         return self._call(list_reviews.sync_detailed, ReviewPage, organization_id, **filters)
 
@@ -622,6 +674,37 @@ async def export_reviews(
         except httpx.HTTPError as error:
             raise _network(error) from None
 
+    async def iter_export_pages(
+        self,
+        organization_id: UUID,
+        dataset: ExportDataset,
+        *,
+        from_time: str,
+        to_time: str,
+        format: ExportFormat = "ndjson",
+        cursor: str | None = None,
+        group_by: str | None = None,
+        max_pages: int = 100,
+    ) -> AsyncIterator[ExportPage]:
+        """Stream export pages, refusing cursor cycles and incomplete bounded traversals."""
+        _export_page_limit(max_pages)
+        seen = {cursor} if cursor is not None else set()
+        for _ in range(max_pages):
+            page = await self.export_reviews(
+                organization_id,
+                dataset,
+                from_time=from_time,
+                to_time=to_time,
+                format=format,
+                cursor=cursor,
+                group_by=group_by,
+            )
+            cursor = _export_next_cursor(page, seen)
+            yield page
+            if cursor is None:
+                return
+        raise _pagination_limit("Export")
+
     async def list_reviews(
         self, organization_id: UUID, **filters: Unpack[ReviewFilters]
     ) -> ReviewPage:

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

tests/test_http.py

@@ -1506,3 +1506,144 @@ async def invoke(name: str, *args: Any, **kwargs: Any) -> Any:
             with pytest.raises(ProoflineError):
                 await invoke("request_reverification", str(REVIEW), [str(REVIEW)], "a" * 40)
             assert len(observed) == 5
+
+
+async def next_export(iterator: Any) -> Any:
+    return await anext(iterator) if hasattr(iterator, "__anext__") else next(iterator)
+
+
+@MODES
+@pytest.mark.parametrize("format", ["csv", "ndjson"])
+async def test_export_iterator_preserves_filters_and_streams_empty_pages(
+    mode: str, format: str
+) -> None:
+    def respond(request: Request, index: int) -> Reply:
+        assert request.path.startswith(f"/api/v1/organizations/{ORG}/exports/finding_outcomes?")
+        assert request.headers["Authorization"] == f"Bearer {TOKEN}"
+        query = parse_qs(urlsplit(request.path).query)
+        assert query == {
+            "from": ["2026-10-01T00:00:00Z"],
+            "to": ["2026-10-25T00:00:00Z"],
+            "format": [format],
+            "group_by": ["repository"],
+            "cursor": ["resume/+cursor" if index == 0 else "next/+cursor"],
+        }
+        headers = {
+            "Content-Type": "text/csv" if format == "csv" else "application/x-ndjson",
+            "X-Proofline-Export-Version": "1",
+            "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z",
+        }
+        if index == 0:
+            headers["X-Proofline-Export-Next-Cursor"] = "next/+cursor"
+        body = (
+            (
+                b"dataset_version,claim\r\n"
+                if index == 0
+                else b"dataset_version,claim\r\n1,hello\r\n"
+            )
+            if format == "csv"
+            else (b"" if index == 0 else b'{"dataset_version":1,"claim":"hello"}\n')
+        )
+        return Reply(body=body, headers=headers)
+
+    with Server(respond) as server:
+        async with session(mode, server.url) as client:
+            iterator = client.iter_export_pages(
+                ORG,
+                "finding_outcomes",
+                from_time="2026-10-01T00:00:00Z",
+                to_time="2026-10-25T00:00:00Z",
+                format=format,
+                cursor="resume/+cursor",
+                group_by="repository",
+                max_pages=2,
+            )
+            assert not server.requests
+            first = await next_export(iterator)
+            assert first.rows == []
+            assert first.next_cursor == "next/+cursor"
+            assert first.version == 1
+            assert first.settled_until == "2026-10-25T00:00:00Z"
+            assert len(server.requests) == 1
+            second = await next_export(iterator)
+            assert second.rows[0]["claim"] == "hello"
+            assert second.next_cursor is None
+            assert second.text.endswith("\n")
+            if mode == "sync":
+                with pytest.raises(StopIteration):
+                    next(iterator)
+            else:
+                with pytest.raises(StopAsyncIteration):
+                    await anext(iterator)
+        assert len(server.requests) == 2
+
+
+@MODES
+@pytest.mark.parametrize("failure", ["empty", "repeat", "resume", "cycle", "bound", "refused"])
+async def test_export_iterator_refuses_invalid_continuations_and_bounds(
+    mode: str, failure: str
+) -> None:
+    def respond(request: Request, index: int) -> Reply:
+        if failure == "refused":
+            return Reply(status=404, body={"detail": "not found"})
+        cursor = {
+            "empty": "",
+            "repeat": "next",
+            "resume": "initial",
+            "cycle": "a" if index % 2 == 0 else "b",
+            "bound": str(index),
+        }[failure]
+        return Reply(
+            body=b'{"dataset_version":1}\n',
+            headers={
+                "Content-Type": "application/x-ndjson",
+                "X-Proofline-Export-Version": "1",
+                "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z",
+                "X-Proofline-Export-Next-Cursor": cursor,
+            },
+        )
+
+    with Server(respond) as server:
+        async with session(mode, server.url) as client:
+            iterator = client.iter_export_pages(
+                ORG,
+                "reviews",
+                from_time="2026-10-01T00:00:00Z",
+                to_time="2026-10-25T00:00:00Z",
+                max_pages=3,
+                cursor="initial" if failure == "resume" else None,
+            )
+            with pytest.raises(ProoflineError) as raised:
+                for _ in range(4):
+                    await next_export(iterator)
+            if failure == "refused":
+                assert raised.value.status == 404
+            else:
+                suffix = "pagination-limit" if failure == "bound" else "invalid-pagination"
+                assert raised.value.type == f"urn:proofline:sdk:{suffix}"
+        expected_requests = {
+            "empty": 1,
+            "repeat": 2,
+            "resume": 1,
+            "cycle": 3,
+            "bound": 3,
+            "refused": 1,
+        }
+        assert len(server.requests) == expected_requests[failure]
+
+
+@MODES
+@pytest.mark.parametrize("bound", [0, -1, 1001, True, 1.5])
+async def test_export_iterator_validates_bound_before_requests(mode: str, bound: Any) -> None:
+    with Server(lambda request, index: Reply(body={})) as server:
+        async with session(mode, server.url) as client:
+            iterator = client.iter_export_pages(
+                ORG,
+                "reviews",
+                from_time="2026-10-01T00:00:00Z",
+                to_time="2026-10-25T00:00:00Z",
+                max_pages=bound,
+            )
+            with pytest.raises(ValueError, match="max_pages"):
+                await next_export(iterator)
+        assert not server.requests