diff --git a/README.md b/README.md index 0a47119..57eb9f1 100644 --- a/README.md +++ b/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 diff --git a/proofline/client.py b/proofline/client.py index de6b525..cfc41b1 100644 --- a/proofline/client.py +++ b/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: diff --git a/scripts/contracts.py b/scripts/contracts.py index e681909..72156b7 100644 --- a/scripts/contracts.py +++ b/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 diff --git a/tests/test_http.py b/tests/test_http.py index eba316e..b13ce83 100644 --- a/tests/test_http.py +++ b/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