feat(outputs): write handoff reports and run history

src/clean_data_export_api/reports.py

@@ -0,0 +1,133 @@
+"""Write deterministic handoff report files for export runs."""
+
+import csv
+import json
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+from pydantic import TypeAdapter
+
+from clean_data_export_api.models import CleanJob, ExportSummary, OutputPaths, RejectedRow
+
+CLEAN_COLUMNS: tuple[str, ...] = (
+    "job_id",
+    "customer_name",
+    "job_status",
+    "work_type",
+    "scheduled_date",
+    "last_updated_at",
+    "source_updated_at",
+)
+
+REJECTED_COLUMNS: tuple[str, ...] = (
+    "source_row_id",
+    "source_job_id",
+    "reason_code",
+    "reason_detail",
+    "raw_record",
+)
+
+_CLEAN_JOBS_ADAPTER = TypeAdapter(list[CleanJob])
+
+
+def output_paths_for(output_dir: Path) -> OutputPaths:
+    """Return the standard output file paths for one export directory."""
+    return OutputPaths(
+        clean_csv=output_dir / "clean_jobs.csv",
+        clean_json=output_dir / "clean_jobs.json",
+        rejected_csv=output_dir / "rejected_jobs.csv",
+        run_summary=output_dir / "run_summary.md",
+    )
+
+
+def write_reports(
+    *,
+    clean_jobs: Iterable[CleanJob],
+    rejected_rows: Iterable[RejectedRow],
+    summary: ExportSummary,
+) -> None:
+    """Write all handoff files for a completed export run."""
+    clean_job_records = tuple(clean_jobs)
+    rejected_row_records = tuple(rejected_rows)
+
+    output_dir = summary.output_paths.clean_csv.parent
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    _write_clean_csv(summary.output_paths.clean_csv, clean_job_records)
+    _write_clean_json(summary.output_paths.clean_json, clean_job_records)
+    _write_rejected_csv(summary.output_paths.rejected_csv, rejected_row_records)
+    _write_run_summary(summary.output_paths.run_summary, summary)
+
+
+def _write_clean_csv(path: Path, clean_jobs: Iterable[CleanJob]) -> None:
+    with path.open("w", newline="", encoding="utf-8") as file:
+        writer = csv.DictWriter(file, fieldnames=CLEAN_COLUMNS)
+        writer.writeheader()
+        for job in clean_jobs:
+            writer.writerow(_clean_job_row(job))
+
+
+def _write_clean_json(path: Path, clean_jobs: Iterable[CleanJob]) -> None:
+    clean_job_payload = _CLEAN_JOBS_ADAPTER.dump_python(list(clean_jobs), mode="json")
+    path.write_text(json.dumps(clean_job_payload, indent=2) + "\n", encoding="utf-8")
+
+
+def _write_rejected_csv(path: Path, rejected_rows: Iterable[RejectedRow]) -> None:
+    with path.open("w", newline="", encoding="utf-8") as file:
+        writer = csv.DictWriter(file, fieldnames=REJECTED_COLUMNS)
+        writer.writeheader()
+        for row in rejected_rows:
+            writer.writerow(_rejected_row(row))
+
+
+def _write_run_summary(path: Path, summary: ExportSummary) -> None:
+    path.write_text(
+        "\n".join(
+            (
+                "# Export Run Summary",
+                "",
+                f"Run ID: {summary.run_id}",
+                f"Requested date range: {summary.from_date} to {summary.to_date}",
+                f"Source records fetched: {summary.source_count}",
+                f"Clean records written: {summary.clean_count}",
+                f"Duplicates removed: {summary.duplicate_count}",
+                f"Rejected records written: {summary.rejected_count}",
+                "",
+                "## Output File Paths",
+                "",
+                f"- Clean CSV: `{summary.output_paths.clean_csv}`",
+                f"- Clean JSON: `{summary.output_paths.clean_json}`",
+                f"- Rejected CSV: `{summary.output_paths.rejected_csv}`",
+                f"- Run summary: `{summary.output_paths.run_summary}`",
+                "",
+                "## Known Limits",
+                "",
+                "- Uses fictional local fixture data only.",
+                "- Covers jobs and job_updates only.",
+                "- Does not connect to a real vendor API.",
+                "",
+                "## Suggested Next Steps",
+                "",
+                "- Review rejected rows before using the clean export.",
+                "- Add buyer-specific validation rules before a real integration.",
+                "- Review vendor API docs before replacing the fixture source.",
+                "",
+            )
+        ),
+        encoding="utf-8",
+    )
+
+
+def _clean_job_row(job: CleanJob) -> dict[str, Any]:
+    return job.model_dump(mode="json")
+
+
+def _rejected_row(row: RejectedRow) -> dict[str, str]:
+    return {
+        "source_row_id": row.source_row_id or "",
+        "source_job_id": row.source_job_id or "",
+        "reason_code": row.reason_code,
+        "reason_detail": row.reason_detail,
+        "raw_record": json.dumps(row.raw_record, sort_keys=True),
+    }

src/clean_data_export_api/repository.py

@@ -0,0 +1,127 @@
+"""Persist export run history in SQLite."""
+
+import sqlite3
+from contextlib import closing
+from datetime import UTC, date, datetime
+from pathlib import Path
+
+from clean_data_export_api.models import OutputPaths
+
+EXPORT_RUN_COLUMNS: tuple[str, ...] = (
+    "id",
+    "started_at",
+    "from_date",
+    "to_date",
+    "source_count",
+    "clean_count",
+    "duplicate_count",
+    "rejected_count",
+    "clean_csv_path",
+    "clean_json_path",
+    "rejected_csv_path",
+    "run_summary_path",
+)
+
+_EXPORT_RUN_VALUE_COLUMNS = EXPORT_RUN_COLUMNS[1:]
+_EXPORT_RUN_COLUMN_DEFINITIONS: tuple[tuple[str, str], ...] = (
+    ("id", "integer primary key autoincrement"),
+    ("started_at", "text not null"),
+    ("from_date", "text not null"),
+    ("to_date", "text not null"),
+    ("source_count", "integer not null"),
+    ("clean_count", "integer not null"),
+    ("duplicate_count", "integer not null"),
+    ("rejected_count", "integer not null"),
+    ("clean_csv_path", "text not null"),
+    ("clean_json_path", "text not null"),
+    ("rejected_csv_path", "text not null"),
+    ("run_summary_path", "text not null"),
+)
+
+_CREATE_EXPORT_RUNS_SQL = (
+    "create table if not exists export_runs ("
+    + ", ".join(f"{name} {definition}" for name, definition in _EXPORT_RUN_COLUMN_DEFINITIONS)
+    + ")"
+)
+_INSERT_EXPORT_RUN_SQL = (
+    f"insert into export_runs ({', '.join(_EXPORT_RUN_VALUE_COLUMNS)}) "
+    f"values ({', '.join('?' for _ in _EXPORT_RUN_VALUE_COLUMNS)})"
+)
+
+
+class RunRepository:
+    """Small SQLite repository for export run history."""
+
+    def __init__(self, database_path: Path) -> None:
+        self._database_path = database_path
+
+    def insert_run(
+        self,
+        *,
+        from_date: date,
+        to_date: date,
+        source_count: int,
+        clean_count: int,
+        duplicate_count: int,
+        rejected_count: int,
+        output_paths: OutputPaths,
+        started_at: datetime | None = None,
+    ) -> int:
+        """Insert one export run and return its run id."""
+        self._database_path.parent.mkdir(parents=True, exist_ok=True)
+        with closing(sqlite3.connect(self._database_path)) as connection:
+            _ensure_schema(connection)
+            cursor = connection.execute(
+                _INSERT_EXPORT_RUN_SQL,
+                _export_run_values(
+                    from_date=from_date,
+                    to_date=to_date,
+                    source_count=source_count,
+                    clean_count=clean_count,
+                    duplicate_count=duplicate_count,
+                    rejected_count=rejected_count,
+                    output_paths=output_paths,
+                    started_at=started_at or datetime.now(UTC),
+                ),
+            )
+            run_id = cursor.lastrowid
+            if run_id is None:
+                msg = "SQLite did not return an export run id"
+                raise RuntimeError(msg)
+            connection.commit()
+            return run_id
+
+
+def _ensure_schema(connection: sqlite3.Connection) -> None:
+    connection.execute(_CREATE_EXPORT_RUNS_SQL)
+
+
+def _export_run_values(
+    *,
+    from_date: date,
+    to_date: date,
+    source_count: int,
+    clean_count: int,
+    duplicate_count: int,
+    rejected_count: int,
+    output_paths: OutputPaths,
+    started_at: datetime,
+) -> tuple[str, str, str, int, int, int, int, str, str, str, str]:
+    return (
+        _format_timestamp(started_at),
+        from_date.isoformat(),
+        to_date.isoformat(),
+        source_count,
+        clean_count,
+        duplicate_count,
+        rejected_count,
+        str(output_paths.clean_csv),
+        str(output_paths.clean_json),
+        str(output_paths.rejected_csv),
+        str(output_paths.run_summary),
+    )
+
+
+def _format_timestamp(value: datetime) -> str:
+    utc_value = value if value.tzinfo is not None else value.replace(tzinfo=UTC)
+    return utc_value.astimezone(UTC).isoformat().replace("+00:00", "Z")

tests/test_reports.py

@@ -0,0 +1,100 @@
+import csv
+import json
+from datetime import UTC, date, datetime
+
+from clean_data_export_api.models import CleanJob, ExportSummary, RejectedRow
+from clean_data_export_api.reports import (
+    CLEAN_COLUMNS,
+    REJECTED_COLUMNS,
+    output_paths_for,
+    write_reports,
+)
+
+
+def test_output_paths_for_uses_documented_filenames(tmp_path) -> None:
+    paths = output_paths_for(tmp_path)
+
+    assert paths.clean_csv == tmp_path / "clean_jobs.csv"
+    assert paths.clean_json == tmp_path / "clean_jobs.json"
+    assert paths.rejected_csv == tmp_path / "rejected_jobs.csv"
+    assert paths.run_summary == tmp_path / "run_summary.md"
+
+
+def test_write_reports_creates_deterministic_handoff_files(tmp_path) -> None:
+    paths = output_paths_for(tmp_path / "nested" / "outputs")
+    clean_jobs = [_clean_job()]
+    rejected_rows = [_rejected_row()]
+    summary = ExportSummary(
+        run_id=1,
+        from_date=date(2026, 6, 1),
+        to_date=date(2026, 6, 5),
+        source_count=9,
+        clean_count=1,
+        duplicate_count=0,
+        rejected_count=1,
+        output_paths=paths,
+    )
+
+    write_reports(clean_jobs=clean_jobs, rejected_rows=rejected_rows, summary=summary)
+
+    with paths.clean_csv.open(newline="", encoding="utf-8") as file:
+        clean_csv_rows = list(csv.DictReader(file))
+    assert clean_csv_rows == [
+        {
+            "job_id": "JOB-1001",
+            "customer_name": "Avery Plumbing",
+            "job_status": "scheduled",
+            "work_type": "repair",
+            "scheduled_date": "2026-06-02",
+            "last_updated_at": "2026-06-02T08:45:00Z",
+            "source_updated_at": "2026-06-01T09:15:00Z",
+        }
+    ]
+    assert tuple(clean_csv_rows[0]) == CLEAN_COLUMNS
+
+    clean_json = json.loads(paths.clean_json.read_text(encoding="utf-8"))
+    assert clean_json == clean_csv_rows
+
+    with paths.rejected_csv.open(newline="", encoding="utf-8") as file:
+        rejected_csv_rows = list(csv.DictReader(file))
+    assert tuple(rejected_csv_rows[0]) == REJECTED_COLUMNS
+    assert rejected_csv_rows == [
+        {
+            "source_row_id": "row-004",
+            "source_job_id": "",
+            "reason_code": "missing_job_id",
+            "reason_detail": "job_id is required",
+            "raw_record": '{"job_id": "", "source_row_id": "row-004"}',
+        }
+    ]
+
+    run_summary = paths.run_summary.read_text(encoding="utf-8")
+    assert "Requested date range: 2026-06-01 to 2026-06-05" in run_summary
+    assert "Source records fetched: 9" in run_summary
+    assert "Clean records written: 1" in run_summary
+    assert "Duplicates removed: 0" in run_summary
+    assert "Rejected records written: 1" in run_summary
+    assert f"Clean CSV: `{paths.clean_csv}`" in run_summary
+    assert "Uses fictional local fixture data only." in run_summary
+
+
+def _clean_job() -> CleanJob:
+    return CleanJob(
+        job_id="JOB-1001",
+        customer_name="Avery Plumbing",
+        job_status="scheduled",
+        work_type="repair",
+        scheduled_date=date(2026, 6, 2),
+        last_updated_at=datetime(2026, 6, 2, 8, 45, tzinfo=UTC),
+        source_updated_at=datetime(2026, 6, 1, 9, 15, tzinfo=UTC),
+    )
+
+
+def _rejected_row() -> RejectedRow:
+    return RejectedRow(
+        source_row_id="row-004",
+        source_job_id=None,
+        reason_code="missing_job_id",
+        reason_detail="job_id is required",
+        raw_record={"source_row_id": "row-004", "job_id": ""},
+    )