feat(export): connect job export service to api and cli

pyproject.toml

@@ -21,7 +21,7 @@ clean-data-export-api = "clean_data_export_api.cli:app"
 [dependency-groups]
 dev = [
     "basedpyright>=1.39.6",
-    "httpx>=0.27.0",
+    "httpx2>=2.3.0",
     "pytest>=8.0.0",
     "pytest-cov>=6.0.0",
     "ruff>=0.8.0",

src/clean_data_export_api/app.py

@@ -0,0 +1,23 @@
+"""FastAPI application for the local clean job export workflow."""
+
+from fastapi import FastAPI
+
+from clean_data_export_api.config import RuntimeSettings, get_default_settings
+from clean_data_export_api.export_service import run_job_export
+from clean_data_export_api.models import ExportJobsRequest, ExportRequest, ExportSummary
+
+
+def create_app(settings: RuntimeSettings | None = None) -> FastAPI:
+    """Create a FastAPI app with local runtime settings."""
+    runtime_settings = settings or get_default_settings()
+    app = FastAPI(title="Clean Data Export API")
+
+    @app.post("/exports/jobs", response_model=ExportSummary)
+    def export_jobs(request: ExportJobsRequest) -> ExportSummary:
+        export_request = ExportRequest(from_date=request.from_date, to_date=request.to_date)
+        return run_job_export(export_request, settings=runtime_settings)
+
+    return app
+
+
+app = create_app()

src/clean_data_export_api/cli.py

@@ -1,13 +1,73 @@
-"""Command shell for the Clean Data Export API project."""
+"""Command line interface for the Clean Data Export API project."""
+
+import json
+from datetime import date
+from pathlib import Path
+from typing import Annotated
 
 import typer
 
+from clean_data_export_api.config import RuntimeSettings
+from clean_data_export_api.export_service import run_job_export
+from clean_data_export_api.models import ExportRequest
+
 app = typer.Typer(
-    help="Command shell for the planned clean job export workflow.",
+    help="Clean job export workflow commands.",
     no_args_is_help=True,
 )
+export_app = typer.Typer(help="Run export workflows.", no_args_is_help=True)
+app.add_typer(export_app, name="export")
 
 
 @app.callback()
 def main() -> None:
     """Show project command help."""
+
+
+@export_app.command("jobs")
+def export_jobs(
+    from_date: Annotated[str, typer.Option("--from-date")],
+    to_date: Annotated[str, typer.Option("--to-date")],
+    sample_data_dir: Annotated[Path, typer.Option("--sample-data-dir")] = Path("sample_data"),
+    output_dir: Annotated[Path, typer.Option("--output-dir")] = Path("outputs"),
+    database_path: Annotated[Path, typer.Option("--database-path")] = Path(
+        "outputs/run_history.sqlite3"
+    ),
+    page_size: Annotated[int, typer.Option("--page-size", min=1)] = 50,
+) -> None:
+    """Export demo job records to clean handoff files."""
+    parsed_from_date = _parse_date_option(from_date, "from_date")
+    parsed_to_date = _parse_date_option(to_date, "to_date")
+    settings = RuntimeSettings(
+        sample_data_dir=sample_data_dir,
+        output_dir=output_dir,
+        database_path=database_path,
+        page_size=page_size,
+    )
+    summary = run_job_export(
+        ExportRequest(
+            from_date=parsed_from_date,
+            to_date=parsed_to_date,
+            output_dir=output_dir,
+        ),
+        settings=settings,
+    )
+    typer.echo(json.dumps(summary.model_dump(mode="json"), indent=2))
+
+
+@app.command()
+def serve(
+    host: Annotated[str, typer.Option("--host")] = "127.0.0.1",
+    port: Annotated[int, typer.Option("--port", min=1, max=65535)] = 8000,
+) -> None:
+    """Serve the export API locally."""
+    import uvicorn
+
+    uvicorn.run("clean_data_export_api.app:app", host=host, port=port, reload=False)
+
+
+def _parse_date_option(value: str, field_name: str) -> date:
+    try:
+        return date.fromisoformat(value)
+    except ValueError as error:
+        raise typer.BadParameter(f"{field_name} must be a valid ISO date") from error

src/clean_data_export_api/export_service.py

@@ -0,0 +1,83 @@
+"""Orchestrate the clean job export workflow."""
+
+from clean_data_export_api.config import RuntimeSettings, get_default_settings
+from clean_data_export_api.mapping import map_source_jobs
+from clean_data_export_api.models import ExportRequest, ExportRunRecord, ExportSummary, SourceJob
+from clean_data_export_api.reports import output_paths_for, write_reports
+from clean_data_export_api.repository import RunRepository
+from clean_data_export_api.source_api import FixtureSourceAPI, SourceJobPage
+from clean_data_export_api.validation import partition_valid_jobs, remove_duplicate_jobs
+
+
+def run_job_export(
+    request: ExportRequest,
+    *,
+    settings: RuntimeSettings | None = None,
+) -> ExportSummary:
+    """Run the full local job export workflow and return its summary."""
+    runtime_settings = settings or get_default_settings()
+    output_dir = request.output_dir or runtime_settings.output_dir
+
+    source_api = FixtureSourceAPI(runtime_settings)
+    source_jobs = _fetch_all_jobs(source_api, request, runtime_settings.page_size)
+    source_job_ids = tuple(job.job_id for job in source_jobs if job.job_id)
+    job_updates = source_api.list_job_updates(source_job_ids)
+
+    candidates = map_source_jobs(source_jobs, job_updates)
+    valid_jobs, validation_rejects = partition_valid_jobs(candidates)
+    clean_jobs, duplicate_rejects = remove_duplicate_jobs(valid_jobs)
+    rejected_rows = (*validation_rejects, *duplicate_rejects)
+
+    output_paths = output_paths_for(output_dir)
+    run_record = ExportRunRecord(
+        from_date=request.from_date,
+        to_date=request.to_date,
+        source_count=len(source_jobs),
+        clean_count=len(clean_jobs),
+        duplicate_count=len(duplicate_rejects),
+        rejected_count=len(rejected_rows),
+        output_paths=output_paths,
+    )
+
+    run_repository = RunRepository(runtime_settings.database_path)
+    run_id = run_repository.insert_run(run_record)
+    summary = ExportSummary(
+        run_id=run_id,
+        from_date=run_record.from_date,
+        to_date=run_record.to_date,
+        source_count=run_record.source_count,
+        clean_count=run_record.clean_count,
+        duplicate_count=run_record.duplicate_count,
+        rejected_count=run_record.rejected_count,
+        output_paths=run_record.output_paths,
+    )
+    try:
+        write_reports(clean_jobs=clean_jobs, rejected_rows=rejected_rows, summary=summary)
+    except Exception as error:
+        try:
+            run_repository.delete_run(run_id)
+        except Exception as cleanup_error:
+            error.add_note(f"Could not remove failed export run {run_id}: {cleanup_error}")
+        raise
+    return summary
+
+
+def _fetch_all_jobs(
+    source_api: FixtureSourceAPI,
+    request: ExportRequest,
+    page_size: int,
+) -> tuple[SourceJob, ...]:
+    records: list[SourceJob] = []
+    cursor = 0
+
+    while True:
+        page: SourceJobPage = source_api.list_jobs(
+            from_date=request.from_date,
+            to_date=request.to_date,
+            cursor=cursor,
+            page_size=page_size,
+        )
+        records.extend(page.items)
+        if page.next_cursor is None:
+            return tuple(records)
+        cursor = page.next_cursor

src/clean_data_export_api/models.py

@@ -4,7 +4,7 @@
 from pathlib import Path
 from typing import Any, Literal, Self
 
-from pydantic import BaseModel, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 RejectedReasonCode = Literal[
     "missing_job_id",
@@ -25,12 +25,13 @@
 )
 
 
-class ExportRequest(BaseModel):
-    """Request contract shared by future API and CLI entry points."""
+class ExportDateRange(BaseModel):
+    """Shared inclusive date range contract for export requests."""
+
+    model_config = ConfigDict(extra="forbid")
 
     from_date: date
     to_date: date
-    output_dir: Path = Path("outputs")
 
     @model_validator(mode="after")
     def validate_date_range(self) -> Self:
@@ -41,6 +42,16 @@ def validate_date_range(self) -> Self:
         return self
 
 
+class ExportJobsRequest(ExportDateRange):
+    """Public API request body for the jobs export endpoint."""
+
+
+class ExportRequest(ExportDateRange):
+    """Internal export request passed to orchestration."""
+
+    output_dir: Path | None = None
+
+
 class OutputPaths(BaseModel):
     """Paths written by a completed export run."""
 
@@ -50,6 +61,18 @@ class OutputPaths(BaseModel):
     run_summary: Path
 
 
+class ExportRunRecord(BaseModel):
+    """Validated export run data persisted before the final summary is returned."""
+
+    from_date: date
+    to_date: date
+    source_count: int = Field(ge=0)
+    clean_count: int = Field(ge=0)
+    duplicate_count: int = Field(ge=0)
+    rejected_count: int = Field(ge=0)
+    output_paths: OutputPaths
+
+
 class ExportSummary(BaseModel):
     """Summary returned by the export service and entry points."""
 

src/clean_data_export_api/repository.py

@@ -2,10 +2,10 @@
 
 import sqlite3
 from contextlib import closing
-from datetime import UTC, date, datetime
+from datetime import UTC, datetime
 from pathlib import Path
 
-from clean_data_export_api.models import OutputPaths
+from clean_data_export_api.models import ExportRunRecord
 
 EXPORT_RUN_COLUMNS: tuple[str, ...] = (
     "id",
@@ -47,6 +47,7 @@
     f"insert into export_runs ({', '.join(_EXPORT_RUN_VALUE_COLUMNS)}) "
     f"values ({', '.join('?' for _ in _EXPORT_RUN_VALUE_COLUMNS)})"
 )
+_DELETE_EXPORT_RUN_SQL = "delete from export_runs where id = ?"
 
 
 class RunRepository:
@@ -57,14 +58,8 @@ def __init__(self, database_path: Path) -> None:
 
     def insert_run(
         self,
+        record: ExportRunRecord,
         *,
-        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."""
@@ -74,13 +69,7 @@ def insert_run(
             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,
+                    record=record,
                     started_at=started_at or datetime.now(UTC),
                 ),
             )
@@ -91,34 +80,35 @@ def insert_run(
             connection.commit()
             return run_id
 
+    def delete_run(self, run_id: int) -> None:
+        """Delete one export run by id."""
+        with closing(sqlite3.connect(self._database_path)) as connection:
+            _ensure_schema(connection)
+            connection.execute(_DELETE_EXPORT_RUN_SQL, (run_id,))
+            connection.commit()
+
 
 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,
+    record: ExportRunRecord,
     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),
+        record.from_date.isoformat(),
+        record.to_date.isoformat(),
+        record.source_count,
+        record.clean_count,
+        record.duplicate_count,
+        record.rejected_count,
+        str(record.output_paths.clean_csv),
+        str(record.output_paths.clean_json),
+        str(record.output_paths.rejected_csv),
+        str(record.output_paths.run_summary),
     )