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