feat(rules): map and validate job export rows

src/clean_data_export_api/mapping.py

@@ -0,0 +1,76 @@
+"""Map source records into clean job export candidates."""
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Any
+
+from clean_data_export_api.models import SourceJob, SourceJobUpdate
+
+
+@dataclass(frozen=True)
+class JobExportCandidate:
+    """Mapped source job before business validation."""
+
+    source_row_id: str
+    job_id: str | None
+    customer_name: str | None
+    job_status: str | None
+    work_type: str | None
+    scheduled_date: str | None
+    source_updated_at: datetime
+    last_updated_at: datetime
+    raw_record: dict[str, Any]
+
+
+def map_source_job(
+    source_job: SourceJob,
+    job_updates: Iterable[SourceJobUpdate],
+) -> JobExportCandidate:
+    """Map one source job and related updates into an export candidate."""
+    source_updated_at = _parse_source_datetime(source_job.updated_at, source_job.source_row_id)
+    related_update_timestamps = tuple(
+        _parse_update_datetime(update.updated_at, update.update_id)
+        for update in job_updates
+        if source_job.job_id is not None and update.job_id == source_job.job_id
+    )
+    last_updated_at = max((source_updated_at, *related_update_timestamps))
+
+    return JobExportCandidate(
+        source_row_id=source_job.source_row_id,
+        job_id=source_job.job_id,
+        customer_name=source_job.customer_name,
+        job_status=source_job.status,
+        work_type=source_job.work_type,
+        scheduled_date=source_job.scheduled_date,
+        source_updated_at=source_updated_at,
+        last_updated_at=last_updated_at,
+        raw_record=source_job.model_dump(mode="json"),
+    )
+
+
+def map_source_jobs(
+    source_jobs: Iterable[SourceJob],
+    job_updates: Iterable[SourceJobUpdate],
+) -> tuple[JobExportCandidate, ...]:
+    """Map source jobs in their existing order."""
+    all_updates = tuple(job_updates)
+    return tuple(map_source_job(source_job, all_updates) for source_job in source_jobs)
+
+
+def _parse_source_datetime(value: str | None, source_row_id: str) -> datetime:
+    if not value:
+        msg = f"source job {source_row_id} is missing updated_at"
+        raise ValueError(msg)
+    return _parse_iso_datetime(value)
+
+
+def _parse_update_datetime(value: str, update_id: str) -> datetime:
+    if not value:
+        msg = f"source job update {update_id} is missing updated_at"
+        raise ValueError(msg)
+    return _parse_iso_datetime(value)
+
+
+def _parse_iso_datetime(value: str) -> datetime:
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))

src/clean_data_export_api/validation.py

@@ -0,0 +1,132 @@
+"""Validate mapped job export candidates and apply duplicate policy."""
+
+from collections.abc import Iterable
+from dataclasses import dataclass
+from datetime import date
+from typing import Any
+
+from clean_data_export_api.mapping import JobExportCandidate
+from clean_data_export_api.models import CleanJob, RejectedRow
+
+VALID_JOB_STATUSES: frozenset[str] = frozenset({"scheduled", "in_progress", "completed"})
+
+
+@dataclass(frozen=True)
+class ValidatedJob:
+    """A clean job with source context needed for later duplicate handling."""
+
+    clean_job: CleanJob
+    source_row_id: str
+    raw_record: dict[str, Any]
+
+
+def validate_job_candidate(candidate: JobExportCandidate) -> ValidatedJob | RejectedRow:
+    """Validate one mapped candidate against job export rules."""
+    if _is_blank(candidate.job_id):
+        return _rejected(candidate, "missing_job_id", "job_id is required")
+    if _is_blank(candidate.customer_name):
+        return _rejected(candidate, "missing_customer_name", "customer_name is required")
+    if _is_blank(candidate.work_type):
+        return _rejected(candidate, "missing_work_type", "work_type is required")
+    if candidate.job_status not in VALID_JOB_STATUSES:
+        return _rejected(candidate, "invalid_status", "job_status must be an allowed value")
+
+    scheduled_date = _parse_scheduled_date(candidate)
+    if scheduled_date is None:
+        return _rejected(candidate, "invalid_date", "scheduled_date must be a valid ISO date")
+
+    clean_job = CleanJob(
+        job_id=_require_string(candidate.job_id),
+        customer_name=_require_string(candidate.customer_name),
+        job_status=_require_string(candidate.job_status),
+        work_type=_require_string(candidate.work_type),
+        scheduled_date=scheduled_date,
+        last_updated_at=candidate.last_updated_at,
+        source_updated_at=candidate.source_updated_at,
+    )
+    return ValidatedJob(
+        clean_job=clean_job,
+        source_row_id=candidate.source_row_id,
+        raw_record=candidate.raw_record,
+    )
+
+
+def partition_valid_jobs(
+    candidates: Iterable[JobExportCandidate],
+) -> tuple[tuple[ValidatedJob, ...], tuple[RejectedRow, ...]]:
+    """Split candidates into valid jobs and validation rejects."""
+    valid_jobs: list[ValidatedJob] = []
+    rejected_rows: list[RejectedRow] = []
+
+    for candidate in candidates:
+        result = validate_job_candidate(candidate)
+        if isinstance(result, RejectedRow):
+            rejected_rows.append(result)
+        else:
+            valid_jobs.append(result)
+
+    return tuple(valid_jobs), tuple(rejected_rows)
+
+
+def remove_duplicate_jobs(
+    valid_jobs: Iterable[ValidatedJob],
+) -> tuple[tuple[CleanJob, ...], tuple[RejectedRow, ...]]:
+    """Accept the first valid job id and reject later valid duplicates."""
+    seen_job_ids: set[str] = set()
+    clean_jobs: list[CleanJob] = []
+    duplicate_rows: list[RejectedRow] = []
+
+    for valid_job in valid_jobs:
+        job_id = valid_job.clean_job.job_id
+        if job_id in seen_job_ids:
+            duplicate_rows.append(
+                RejectedRow(
+                    source_row_id=valid_job.source_row_id,
+                    source_job_id=job_id,
+                    reason_code="duplicate_record",
+                    reason_detail=f"duplicate job_id {job_id} was skipped",
+                    raw_record=valid_job.raw_record,
+                )
+            )
+            continue
+
+        seen_job_ids.add(job_id)
+        clean_jobs.append(valid_job.clean_job)
+
+    return tuple(clean_jobs), tuple(duplicate_rows)
+
+
+def _rejected(
+    candidate: JobExportCandidate,
+    reason_code: str,
+    reason_detail: str,
+) -> RejectedRow:
+    return RejectedRow.model_validate(
+        {
+            "source_row_id": candidate.source_row_id,
+            "source_job_id": candidate.job_id,
+            "reason_code": reason_code,
+            "reason_detail": reason_detail,
+            "raw_record": candidate.raw_record,
+        }
+    )
+
+
+def _parse_scheduled_date(candidate: JobExportCandidate) -> date | None:
+    if _is_blank(candidate.scheduled_date):
+        return None
+    try:
+        return date.fromisoformat(_require_string(candidate.scheduled_date))
+    except ValueError:
+        return None
+
+
+def _is_blank(value: str | None) -> bool:
+    return value is None or value.strip() == ""
+
+
+def _require_string(value: str | None) -> str:
+    if value is None:
+        msg = "expected non-null string after validation"
+        raise TypeError(msg)
+    return value

tests/test_duplicate_handling.py

@@ -0,0 +1,93 @@
+from datetime import UTC, datetime
+
+from clean_data_export_api.mapping import JobExportCandidate
+from clean_data_export_api.validation import (
+    partition_valid_jobs,
+    remove_duplicate_jobs,
+)
+
+
+def test_remove_duplicate_jobs_keeps_first_valid_job_id() -> None:
+    valid_jobs, validation_rejects = partition_valid_jobs(
+        [
+            _candidate(source_row_id="row-001", job_id="JOB-1001"),
+            _candidate(source_row_id="row-002", job_id="JOB-1002"),
+            _candidate(source_row_id="row-003", job_id="JOB-1001"),
+        ]
+    )
+
+    clean_jobs, duplicate_rows = remove_duplicate_jobs(valid_jobs)
+
+    assert validation_rejects == ()
+    assert [job.job_id for job in clean_jobs] == ["JOB-1001", "JOB-1002"]
+    assert len(duplicate_rows) == 1
+    assert duplicate_rows[0].source_row_id == "row-003"
+    assert duplicate_rows[0].source_job_id == "JOB-1001"
+    assert duplicate_rows[0].reason_code == "duplicate_record"
+    assert duplicate_rows[0].raw_record["source_row_id"] == "row-003"
+
+
+def test_invalid_rows_are_rejected_before_duplicate_handling() -> None:
+    valid_jobs, validation_rejects = partition_valid_jobs(
+        [
+            _candidate(source_row_id="row-001", job_id="JOB-1001", job_status="waiting"),
+            _candidate(source_row_id="row-002", job_id="JOB-1001"),
+        ]
+    )
+
+    clean_jobs, duplicate_rows = remove_duplicate_jobs(valid_jobs)
+
+    assert [row.reason_code for row in validation_rejects] == ["invalid_status"]
+    assert [job.job_id for job in clean_jobs] == ["JOB-1001"]
+    assert duplicate_rows == ()
+
+
+def test_full_fixture_rule_counts_match_mvp_contract() -> None:
+    candidates = [
+        _candidate(source_row_id="row-001", job_id="JOB-1001"),
+        _candidate(source_row_id="row-002", job_id="JOB-1002"),
+        _candidate(source_row_id="row-003", job_id="JOB-1003"),
+        _candidate(source_row_id="row-004", job_id=None),
+        _candidate(source_row_id="row-005", job_id="JOB-1005", customer_name=""),
+        _candidate(source_row_id="row-006", job_id="JOB-1006", job_status="waiting"),
+        _candidate(source_row_id="row-007", job_id="JOB-1007", scheduled_date="2026-15-01"),
+        _candidate(source_row_id="row-008", job_id="JOB-1008", work_type=None),
+        _candidate(source_row_id="row-009", job_id="JOB-1002"),
+    ]
+
+    valid_jobs, validation_rejects = partition_valid_jobs(candidates)
+    clean_jobs, duplicate_rows = remove_duplicate_jobs(valid_jobs)
+
+    assert len(valid_jobs) == 4
+    assert len(validation_rejects) == 5
+    assert len(clean_jobs) == 3
+    assert len(duplicate_rows) == 1
+    assert [row.reason_code for row in validation_rejects] == [
+        "missing_job_id",
+        "missing_customer_name",
+        "invalid_status",
+        "invalid_date",
+        "missing_work_type",
+    ]
+
+
+def _candidate(
+    *,
+    source_row_id: str,
+    job_id: str | None,
+    customer_name: str | None = "Avery Plumbing",
+    job_status: str | None = "scheduled",
+    work_type: str | None = "repair",
+    scheduled_date: str | None = "2026-06-02",
+) -> JobExportCandidate:
+    return JobExportCandidate(
+        source_row_id=source_row_id,
+        job_id=job_id,
+        customer_name=customer_name,
+        job_status=job_status,
+        work_type=work_type,
+        scheduled_date=scheduled_date,
+        source_updated_at=datetime(2026, 6, 1, 9, 15, tzinfo=UTC),
+        last_updated_at=datetime(2026, 6, 2, 8, 45, tzinfo=UTC),
+        raw_record={"source_row_id": source_row_id, "job_id": job_id},
+    )

tests/test_mapping.py

@@ -0,0 +1,95 @@
+from datetime import UTC, datetime
+
+import pytest
+
+from clean_data_export_api.mapping import map_source_job, map_source_jobs
+from clean_data_export_api.models import SourceJob, SourceJobUpdate
+
+
+def test_map_source_job_uses_latest_job_or_update_timestamp() -> None:
+    source_job = _source_job(updated_at="2026-06-01T09:15:00Z")
+    updates = (
+        _source_update("update-001", "JOB-1001", "2026-06-01T14:00:00Z"),
+        _source_update("update-002", "JOB-1001", "2026-06-02T08:45:00Z"),
+        _source_update("update-003", "JOB-9999", "2026-06-05T12:00:00Z"),
+    )
+
+    candidate = map_source_job(source_job, updates)
+
+    assert candidate.source_row_id == "row-001"
+    assert candidate.job_id == "JOB-1001"
+    assert candidate.customer_name == "Avery Plumbing"
+    assert candidate.job_status == "scheduled"
+    assert candidate.work_type == "repair"
+    assert candidate.scheduled_date == "2026-06-02"
+    assert candidate.source_updated_at == datetime(2026, 6, 1, 9, 15, tzinfo=UTC)
+    assert candidate.last_updated_at == datetime(2026, 6, 2, 8, 45, tzinfo=UTC)
+    assert candidate.raw_record["source_row_id"] == "row-001"
+
+
+def test_map_source_job_uses_source_timestamp_when_no_related_update_is_newer() -> None:
+    source_job = _source_job(updated_at="2026-06-03T09:15:00Z")
+    updates = (_source_update("update-001", "JOB-1001", "2026-06-01T14:00:00Z"),)
+
+    candidate = map_source_job(source_job, updates)
+
+    assert candidate.last_updated_at == datetime(2026, 6, 3, 9, 15, tzinfo=UTC)
+
+
+def test_map_source_jobs_preserves_source_job_order() -> None:
+    source_jobs = (
+        _source_job(source_row_id="row-001", job_id="JOB-1001"),
+        _source_job(source_row_id="row-002", job_id="JOB-1002"),
+    )
+
+    candidates = map_source_jobs(source_jobs, ())
+
+    assert [candidate.source_row_id for candidate in candidates] == ["row-001", "row-002"]
+
+
+def test_map_source_job_rejects_missing_source_updated_at() -> None:
+    source_job = _source_job(updated_at=None)
+
+    with pytest.raises(ValueError, match="source job row-001 is missing updated_at"):
+        map_source_job(source_job, ())
+
+
+def test_map_source_job_rejects_missing_update_updated_at() -> None:
+    source_job = _source_job()
+    update = SourceJobUpdate(
+        update_id="update-001",
+        job_id="JOB-1001",
+        status="scheduled",
+        note="Missing timestamp.",
+        updated_at="",
+    )
+
+    with pytest.raises(ValueError, match="source job update update-001 is missing updated_at"):
+        map_source_job(source_job, (update,))
+
+
+def _source_job(
+    *,
+    source_row_id: str = "row-001",
+    job_id: str | None = "JOB-1001",
+    updated_at: str | None = "2026-06-01T09:15:00Z",
+) -> SourceJob:
+    return SourceJob(
+        source_row_id=source_row_id,
+        job_id=job_id,
+        customer_name="Avery Plumbing",
+        status="scheduled",
+        work_type="repair",
+        scheduled_date="2026-06-02",
+        updated_at=updated_at,
+    )
+
+
+def _source_update(update_id: str, job_id: str, updated_at: str) -> SourceJobUpdate:
+    return SourceJobUpdate(
+        update_id=update_id,
+        job_id=job_id,
+        status="scheduled",
+        note="Fixture update.",
+        updated_at=updated_at,
+    )