From 5cb10d4a0a8b3e8ce684ec6eaa6f8bc7d0b0b2af Mon Sep 17 00:00:00 2001 From: Darri Steinn Konradsson Date: Wed, 4 Nov 2026 10:35:40 +0000 Subject: [PATCH] Iterate export pages with bounded continuation cursors export_reviews read one page and left the continuation loop to the caller. iter_export_pages follows X-Proofline-Export-Next-Cursor on both clients with the same dataset, window, format and grouping, bounded by max_pages (1..1000, default 100). A repeated or empty cursor is refused as invalid pagination, a settled horizon that changes between pages as an invalid response, and an exhausted bound raises pagination-limit instead of stopping silently. A cursor resumes an interrupted export. Export windows accept timezone-aware datetimes as well as RFC3339 strings; naive datetimes raise ValueError before any request, as does an empty cursor on either method. Both facades build the export request through one helper, so dataset, format, cursor and window validation live in one place, and ExportTime joins the public exports. The README keeps the single-page example and documents the iterator below it. --- README.md | 19 +++++ proofline/__init__.py | 3 +- proofline/client.py | 179 ++++++++++++++++++++++++++++----------- proofline/exports.py | 57 +++++++++++-- scripts/contracts.py | 2 +- tests/test_http.py | 190 +++++++++++++++++++++++++++++++++++++++++- 6 files changed, 393 insertions(+), 57 deletions(-) diff --git a/README.md b/README.md index 0a47119..28937ba 100644 --- a/README.md +++ b/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 diff --git a/proofline/__init__.py b/proofline/__init__.py index f9dbdb9..0c81685 100644 --- a/proofline/__init__.py +++ b/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", diff --git a/proofline/client.py b/proofline/client.py index de6b525..a61e280 100644 --- a/proofline/client.py +++ b/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." + ) diff --git a/proofline/exports.py b/proofline/exports.py index a274dd2..96d871e 100644 --- a/proofline/exports.py +++ b/proofline/exports.py @@ -1,13 +1,15 @@ -"""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 @@ -15,6 +17,9 @@ 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 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..51d1f54 100644 --- a/tests/test_http.py +++ b/tests/test_http.py @@ -11,7 +11,7 @@ import traceback from collections.abc import AsyncIterator from contextlib import asynccontextmanager -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime, timedelta, timezone from typing import Any from urllib.parse import parse_qs, urlsplit @@ -28,6 +28,8 @@ CustomerAssessmentDetail, CustomerFinding, CustomerReviewDetail, + ExportDataset, + ExportPage, FileReviewRequest, FileReviewResponse, PatchReviewRequest, @@ -71,6 +73,7 @@ Client = ProoflineClient | AsyncProoflineClient MODES = pytest.mark.parametrize("mode", ["sync", "async"]) +HORIZON = "2026-10-25T00:00:00Z" @asynccontextmanager @@ -95,6 +98,44 @@ async def collect(client: Client, **kwargs: Any) -> list[Any]: return list(iterator) +async def collect_exports( + client: Client, pages: list[ExportPage], dataset: ExportDataset = "findings", **kwargs: Any +) -> None: + window: dict[str, Any] = {"from_time": "2026-10-01T00:00:00Z", "to_time": HORIZON} + iterator = client.iter_export_pages(ORG, dataset, **{**window, **kwargs}) + if hasattr(iterator, "__aiter__"): + async for page in iterator: + pages.append(page) + else: + for page in iterator: + pages.append(page) + + +def export_reply( + index: int, + *, + cursor: str | None = None, + horizon: str = HORIZON, + status: int = 200, + is_csv: bool = False, +) -> Reply: + if status != 200: + return Reply(status=status, body={"detail": "not found"}) + headers = { + "Content-Type": "text/csv" if is_csv else "application/x-ndjson", + "X-Proofline-Export-Version": "1", + "X-Proofline-Export-Settled-Until": horizon, + } + if cursor is not None: + headers["X-Proofline-Export-Next-Cursor"] = cursor + body = ( + f"dataset_version,page\r\n1,{index}\r\n" + if is_csv + else f'{{"dataset_version":1,"page":{index}}}\n' + ) + return Reply(body=body.encode(), headers=headers) + + class Clock: def __init__(self, monkeypatch: pytest.MonkeyPatch) -> None: self.now = 100.0 @@ -1392,6 +1433,153 @@ def respond(request: Request, index: int) -> Reply: assert caught.value.status == 404 +@MODES +async def test_export_iteration_repeats_the_query_until_the_last_page(mode: str) -> None: + 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 request.headers["Authorization"] == f"Bearer {TOKEN}" + assert request.headers["Accept"] == "text/csv" + assert query["from"] == ["2026-10-01T00:00:00Z"] + assert query["to"] == ["2026-10-25T00:00:00.500000Z"] + assert query["format"] == ["csv"] + assert query["group_by"] == ["severity"] + assert query.get("cursor") == (["opaque/+cursor"] if index == 1 else None) + return export_reply(index, cursor="opaque/+cursor" if index == 0 else None, is_csv=True) + + with Server(respond) as server: + async with session(mode, server.url) as client: + pages: list[ExportPage] = [] + await collect_exports( + client, + pages, + "finding_outcomes", + from_time=datetime(2026, 10, 1, 2, tzinfo=timezone(timedelta(hours=2))), + to_time=datetime(2026, 10, 25, 0, 0, 0, 500_000, tzinfo=UTC), + format="csv", + group_by="severity", + ) + assert [page.rows[0]["page"] for page in pages] == ["0", "1"] + assert [page.text.count("dataset_version,page") for page in pages] == [1, 1] + assert [page.settled_until for page in pages] == [HORIZON, HORIZON] + assert pages[1].next_cursor is None + assert len(server.requests) == 2 + assert "cursor=opaque%2F%2Bcursor" in server.requests[1].path + + +@MODES +async def test_export_iteration_refuses_cycles_bounds_horizon_drift_and_problems( + mode: str, +) -> None: + with Server(lambda request, index: export_reply(index)) as server: + async with session(mode, server.url) as client: + pages: list[ExportPage] = [] + await collect_exports(client, pages) + assert len(pages) == 1 + assert len(server.requests) == 1 + with Server(lambda request, index: export_reply(index, cursor="repeated")) as server: + async with session(mode, server.url) as client: + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages) + assert raised.value.type == "urn:proofline:sdk:invalid-pagination" + assert len(pages) == 2 + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages, cursor="repeated") + assert raised.value.type == "urn:proofline:sdk:invalid-pagination" + assert len(pages) == 1 + queries = [parse_qs(urlsplit(request.path).query) for request in server.requests] + assert [query.get("cursor") for query in queries] == [None, ["repeated"], ["repeated"]] + with Server(lambda request, index: export_reply(index, cursor="")) as server: + async with session(mode, server.url) as client: + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages) + assert raised.value.type == "urn:proofline:sdk:invalid-pagination" + assert len(pages) == 1 + assert len(server.requests) == 1 + with Server(lambda request, index: export_reply(index, cursor=str(index))) as server: + async with session(mode, server.url) as client: + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages, max_pages=2) + assert raised.value.type == "urn:proofline:sdk:pagination-limit" + assert len(pages) == 2 + assert len(server.requests) == 2 + drifted = "2026-10-25T00:05:00Z" + with Server( + lambda request, index: export_reply( + index, cursor=str(index), horizon=HORIZON if index == 0 else drifted + ) + ) as server: + async with session(mode, server.url) as client: + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages) + assert raised.value.type == "urn:proofline:sdk:invalid-response" + assert len(pages) == 1 + assert len(server.requests) == 2 + with Server( + lambda request, index: export_reply(index, cursor="next", status=200 if index == 0 else 404) + ) as server: + async with session(mode, server.url) as client: + pages = [] + with pytest.raises(ProoflineError) as raised: + await collect_exports(client, pages) + assert raised.value.status == 404 + assert len(pages) == 1 + assert len(server.requests) == 2 + + +@MODES +async def test_export_windows_serialize_aware_datetimes_and_refuse_bad_arguments( + mode: str, +) -> None: + def respond(request: Request, index: int) -> Reply: + query = parse_qs(urlsplit(request.path).query) + assert query["from"] == ["2026-10-01T00:00:00Z"] + assert query["to"] == ["2026-10-24T22:00:00Z"] + return export_reply(index) + + aware = datetime(2026, 10, 1, tzinfo=UTC) + with Server(respond) as server: + async with session(mode, server.url) as client: + page = await call( + client, + "export_reviews", + ORG, + "findings", + from_time=aware, + to_time=datetime(2026, 10, 25, tzinfo=timezone(timedelta(hours=2))), + ) + assert page.settled_until == HORIZON + assert len(server.requests) == 1 + with Server(lambda request, index: export_reply(index)) as server: + async with session(mode, server.url) as client: + for max_pages in [0, 1_001]: + with pytest.raises(ValueError): + await collect_exports(client, [], max_pages=max_pages) + with pytest.raises(ValueError): + await collect_exports(client, [], cursor="") + naive = datetime(2026, 10, 1) + with pytest.raises(ValueError): + await collect_exports(client, [], from_time=naive) + with pytest.raises(ValueError): + await collect_exports(client, [], to_time=naive) + refusals: list[tuple[str, dict[str, Any]]] = [ + ("nonsense", {}), + ("findings", {"format": "xml"}), + ("findings", {"cursor": ""}), + ("findings", {"from_time": naive}), + ] + for dataset, refused in refusals: + query = {"from_time": aware, "to_time": HORIZON, **refused} + with pytest.raises(ValueError): + await call(client, "export_reviews", ORG, dataset, **query) + assert server.requests == [] + + @MODES async def test_knowledge_negotiation_scope_and_no_retry(mode: str) -> None: refusal = False