Add ExportWriter for combining CSV and NDJSON export pages

README.md

@@ -47,6 +47,24 @@ with ProoflineClient() as client:
     # Continue with cursor=page.next_cursor when it is not None.
 ```
 
+Use `ExportWriter` to combine pages in a text stream. It writes CSV headers once
+and refuses changes to the format or column order before writing that page.
+It retains the original record content and separates records across pages,
+including when a page has no final newline. The caller owns and closes the stream.
+
+```python
+from proofline import ExportWriter
+
+with open("reviews.csv", "w", encoding="utf-8", newline="") as stream:
+    writer = ExportWriter(stream)
+    writer.write_page(page)
+    # Fetch the next page with the same filters, then call write_page again.
+```
+
+`ExportWriter` also accepts pages fetched with `AsyncProoflineClient`. Writing to
+the text stream is synchronous. A write error propagates to the caller; use a
+fresh output file when restarting an interrupted export.
+
 ## Coverage uploads
 
 Use a repository-restricted token with `coverage:write` and current repository write

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, ExportWriter
 from .outcomes import AssessmentGap, CheckGap, ReviewOutcome
 
 __all__ = [
@@ -51,6 +51,7 @@
     "ExportDataset",
     "ExportFormat",
     "ExportPage",
+    "ExportWriter",
     "FileReviewRequest",
     "FileReviewResponse",
     "PatchReviewRequest",

proofline/exports.py

@@ -7,7 +7,7 @@
 import json
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Any, Literal
+from typing import Any, Literal, TextIO
 
 import httpx
 
@@ -27,6 +27,46 @@ class ExportPage:
     next_cursor: str | None
 
 
+class ExportWriter:
+    """Write consecutive export pages to a caller owned text stream.
+
+    CSV headers are emitted once. Format and CSV schema changes are refused
+    before the offending page is written. The stream remains open.
+    """
+
+    def __init__(self, stream: TextIO) -> None:
+        self._stream = stream
+        self._format: ExportFormat | None = None
+        self._columns: list[str] | None = None
+
+    def write_page(self, page: ExportPage) -> None:
+        """Append a page, omitting repeated headers and adding a final newline if needed."""
+        if page.format not in ("csv", "ndjson"):
+            raise ValueError("Export format must be csv or ndjson.")
+        if self._format is not None and page.format != self._format:
+            raise ValueError("Export pages must use the same format.")
+        text = page.text
+        columns = None
+        if page.format == "csv":
+            source = io.StringIO(text, newline="")
+            try:
+                columns = next(csv.reader(source, strict=True))
+            except (StopIteration, csv.Error):
+                raise ValueError("CSV export page must contain a valid header.") from None
+            if not columns or len(set(columns)) != len(columns):
+                raise ValueError("CSV export page must contain unique column names.")
+            if self._columns is not None:
+                if columns != self._columns:
+                    raise ValueError("CSV export pages must use the same columns in order.")
+                text = source.read()
+        if text and not text.endswith("\n" if page.format == "ndjson" else ("\n", "\r")):
+            # A final record may have no newline. Separate the next page.
+            text += "\n"
+        self._stream.write(text)
+        self._format = page.format
+        self._columns = columns
+
+
 def decode_export(response: httpx.Response, format: ExportFormat) -> ExportPage:
     try:
         media = "text/csv" if format == "csv" else "application/x-ndjson"

tests/test_exports.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import csv
+import io
+
+import httpx
+import pytest
+
+from proofline import ExportWriter
+from proofline.exports import ExportFormat, ExportPage, decode_export
+
+
+def export(text: str, format: ExportFormat = "csv") -> ExportPage:
+    return decode_export(
+        httpx.Response(
+            200,
+            content=text.encode(),
+            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",
+            },
+        ),
+        format,
+    )
+
+
+def test_csv_pages_keep_one_header_and_original_multiline_content() -> None:
+    stream = io.StringIO(newline="")
+    writer = ExportWriter(stream)
+    header = 'dataset_version,"claim\ntext"\r\n'
+    first = header + '1,"comma, quote "" and\nnewline"\r\n'
+    second = header + '1,"second"\r\n'
+    writer.write_page(export(first))
+    writer.write_page(export(header))
+    writer.write_page(export(second))
+    assert stream.getvalue() == first + '1,"second"\r\n'
+    assert len(list(csv.reader(io.StringIO(stream.getvalue())))) == 3
+    assert not stream.closed
+
+
+def test_ndjson_pages_without_final_newline_remain_separate_records() -> None:
+    stream = io.StringIO()
+    writer = ExportWriter(stream)
+    writer.write_page(export('{"dataset_version":1}', "ndjson"))
+    writer.write_page(export("", "ndjson"))
+    writer.write_page(export('{"dataset_version":1}\n', "ndjson"))
+    assert stream.getvalue() == '{"dataset_version":1}\n{"dataset_version":1}\n'
+
+
+@pytest.mark.parametrize("header", ["claim,dataset_version\n", "dataset_version,other\n"])
+def test_changed_schema_is_refused_before_writing(header: str) -> None:
+    stream = io.StringIO()
+    writer = ExportWriter(stream)
+    writer.write_page(export("dataset_version,claim\n"))
+    with pytest.raises(ValueError, match="same columns"):
+        writer.write_page(export(header))
+    assert stream.getvalue() == "dataset_version,claim\n"
+    writer.write_page(export("dataset_version,claim\n1,ok\n"))
+    assert stream.getvalue().endswith("1,ok\n")
+
+
+def test_changed_format_is_refused_before_writing() -> None:
+    stream = io.StringIO()
+    writer = ExportWriter(stream)
+    writer.write_page(export("", "ndjson"))
+    with pytest.raises(ValueError, match="same format"):
+        writer.write_page(export("dataset_version,claim\n"))
+    assert stream.getvalue() == ""
+
+
+def test_csv_pages_without_final_newline_remain_separate_records() -> None:
+    stream = io.StringIO(newline="")
+    writer = ExportWriter(stream)
+    writer.write_page(export("dataset_version,claim\n1,first"))
+    writer.write_page(export("dataset_version,claim\n1,second"))
+    assert stream.getvalue() == "dataset_version,claim\n1,first\n1,second\n"
+
+
+def test_ndjson_lone_carriage_return_gets_line_feed() -> None:
+    stream = io.StringIO(newline="")
+    writer = ExportWriter(stream)
+    writer.write_page(export('{"dataset_version":1}\r', "ndjson"))
+    writer.write_page(export('{"dataset_version":1}\n', "ndjson"))
+    assert stream.getvalue() == '{"dataset_version":1}\r\n{"dataset_version":1}\n'