From 960b70d8ece886afaeab80216cb2d9caa44edec7 Mon Sep 17 00:00:00 2001 From: Ramiro Carnicer Souble <104595014+ram4-dev@users.noreply.github.com> Date: Wed, 4 Nov 2026 20:07:36 -0300 Subject: [PATCH] Add sync and async export page iteration --- README.md | 23 ++++++++++ proofline/client.py | 102 ++++++++++++++++++++++++++++++++++++++++++- scripts/contracts.py | 2 +- tests/test_http.py | 88 +++++++++++++++++++++++++++++++++++++ 4 files changed, 212 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index dbb1716..80c413a 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,29 @@ with ProoflineClient() as client: # Continue with cursor=page.next_cursor when it is not None. ``` +`iter_export_pages` runs that loop for one window and yields every page while it +follows the signed cursors, keeping the dataset, window, format and grouping fixed. +It stops at the last page, refuses a cursor it has already received and raises +`ProoflineError` when `max_pages` (100 by default, at most 1000) is reached. +`AsyncProoflineClient` exposes the same helper for `async for`. + +```python +import os +from uuid import UUID + +with ProoflineClient() as client: + rows = [ + row + for export in client.iter_export_pages( + UUID(os.environ["PROOFLINE_ORGANIZATION_ID"]), + "findings", + from_time="2026-10-01T00:00:00Z", + to_time="2026-10-25T00:00:00Z", + ) + for row in export.rows + ] +``` + ## 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..f51d8ed 100644 --- a/proofline/client.py +++ b/proofline/client.py @@ -180,12 +180,34 @@ def _next_cursor(page: ReviewPage, seen: set[str]) -> str | None: return cursor -def _pagination_limit() -> ProoflineError: +def _pagination_limit(subject: str = "Review") -> ProoflineError: return ProoflineError( - 0, "urn:proofline:sdk:pagination-limit", "Review pagination reached the configured bound." + 0, + "urn:proofline:sdk:pagination-limit", + f"{subject} pagination reached the configured bound.", ) +def _export_limits(max_pages: int) -> None: + if not 1 <= max_pages <= 1_000: + raise ValueError("Export iteration requires max_pages 1..1000.") + + +def _export_cursor(page: ExportPage, seen: set[str]) -> str | None: + """A missing continuation ends the window; a repeated one is refused.""" + cursor = page.next_cursor + if not cursor: + return None + if 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. @@ -491,6 +513,44 @@ def iter_reviews( params["cursor"] = cursor raise _pagination_limit() + def iter_export_pages( + self, + organization_id: UUID, + dataset: ExportDataset, + *, + from_time: str, + to_time: str, + format: ExportFormat = "ndjson", + group_by: str | None = None, + cursor: str | None = None, + max_pages: int = 100, + ) -> Iterator[ExportPage]: + """Read every export page of one window by following signed cursors. + + The window, format and grouping stay fixed across the requests, so a + continuation only changes the cursor. Iteration stops at the last page, + refuses a repeated cursor and stops at the configured page bound. + """ + _export_limits(max_pages) + seen: set[str] = set() + if cursor: + seen.add(cursor) + 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, + ) + yield page + cursor = _export_cursor(page, seen) + if cursor is None: + return + raise _pagination_limit("Export") + def wait_for_review( self, organization_id: UUID, review_id: UUID, *, timeout: float = 300 ) -> ReviewOutcome: @@ -850,6 +910,44 @@ async def iter_reviews( params["cursor"] = cursor raise _pagination_limit() + async def iter_export_pages( + self, + organization_id: UUID, + dataset: ExportDataset, + *, + from_time: str, + to_time: str, + format: ExportFormat = "ndjson", + group_by: str | None = None, + cursor: str | None = None, + max_pages: int = 100, + ) -> AsyncIterator[ExportPage]: + """Read every export page of one window by following signed cursors. + + The window, format and grouping stay fixed across the requests, so a + continuation only changes the cursor. Iteration stops at the last page, + refuses a repeated cursor and stops at the configured page bound. + """ + _export_limits(max_pages) + seen: set[str] = set() + if cursor: + seen.add(cursor) + 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, + ) + yield page + cursor = _export_cursor(page, seen) + if cursor is None: + return + raise _pagination_limit("Export") + async def wait_for_review( self, organization_id: UUID, review_id: UUID, *, timeout: float = 300 ) -> ReviewOutcome: 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..cbf1544 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -28,6 +28,7 @@ CustomerAssessmentDetail, CustomerFinding, CustomerReviewDetail, + ExportPage, FileReviewRequest, FileReviewResponse, PatchReviewRequest, @@ -95,6 +96,13 @@ async def collect(client: Client, **kwargs: Any) -> list[Any]: return list(iterator) +async def collect_pages(client: Client, dataset: str, **kwargs: Any) -> list[ExportPage]: + iterator = client.iter_export_pages(ORG, dataset, **kwargs) + if hasattr(iterator, "__aiter__"): + return [export async for export in iterator] + return list(iterator) + + class Clock: def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None: self.now = 100.0 @@ -1392,6 +1400,86 @@ def respond(request: Request, index: int) -> Reply: assert caught.value.status == 404 +@MODES +async def test_export_iteration_keeps_the_window_and_refuses_cycles_or_bounds(mode: str) -> None: + kwargs = { + "from_time": "2026-10-01T00:00:00Z", + "to_time": "2026-10-25T00:00:00Z", + "format": "csv", + "group_by": "severity", + } + + def csv_reply(cursor: str | None) -> Reply: + headers = { + "Content-Type": "text/csv", + "X-Proofline-Export-Version": "1", + "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z", + } + if cursor is not None: + headers["X-Proofline-Export-Next-Cursor"] = cursor + return Reply(body=b"dataset_version,severity\r\n1,high\r\n", headers=headers) + + def respond(request: Request, index: int) -> Reply: + query = parse_qs(urlsplit(request.path).query) + assert request.path.startswith(f"/api/v1/organizations/{ORG}/exports/finding_outcomes?") + assert query["from"] == ["2026-10-01T00:00:00Z"] + assert query["to"] == ["2026-10-25T00:00:00Z"] + assert query["format"] == ["csv"] + assert query["group_by"] == ["severity"] + assert query.get("cursor") == ([f"cursor-{index - 1}"] if index else None) + headers = { + "Content-Type": "text/csv", + "X-Proofline-Export-Version": "1", + "X-Proofline-Export-Settled-Until": "2026-10-25T00:00:00Z", + } + if index < 2: + headers["X-Proofline-Export-Next-Cursor"] = f"cursor-{index}" + return Reply( + body=f"dataset_version,severity\r\n1,page-{index}\r\n".encode(), headers=headers + ) + + with Server(respond) as server: + async with session(mode, server.url) as client: + exports = await collect_pages(client, "finding_outcomes", **kwargs) + assert [export.rows[0]["severity"] for export in exports] == [ + "page-0", + "page-1", + "page-2", + ] + assert exports[0].text.startswith("dataset_version,severity\r\n") + assert [export.next_cursor for export in exports] == ["cursor-0", "cursor-1", None] + assert exports[-1].settled_until == "2026-10-25T00:00:00Z" + assert len(server.requests) == 3 + + with Server(lambda request, index: csv_reply("repeated")) as server: + async with session(mode, server.url) as client: + with pytest.raises(ProoflineError) as raised: + await collect_pages(client, "finding_outcomes", **kwargs) + assert raised.value.type == "urn:proofline:sdk:invalid-pagination" + assert len(server.requests) == 2 + + with Server(lambda request, index: csv_reply("seeded")) as server: + async with session(mode, server.url) as client: + with pytest.raises(ProoflineError) as raised: + await collect_pages(client, "finding_outcomes", cursor="seeded", **kwargs) + assert raised.value.type == "urn:proofline:sdk:invalid-pagination" + assert len(server.requests) == 1 + + with Server(lambda request, index: csv_reply(f"cursor-{index}")) as server: + async with session(mode, server.url) as client: + with pytest.raises(ProoflineError) as raised: + await collect_pages(client, "finding_outcomes", max_pages=2, **kwargs) + assert raised.value.detail == "Export pagination reached the configured bound." + assert len(server.requests) == 2 + + with Server(lambda request, index: csv_reply(None)) as server: + async with session(mode, server.url) as client: + for bound in (0, 1_001): + with pytest.raises(ValueError): + await collect_pages(client, "finding_outcomes", max_pages=bound, **kwargs) + assert server.requests == [] + + @MODES async def test_knowledge_negotiation_scope_and_no_retry(mode: str) -> None: refusal = False