From 400398755befaa314bfa24bc3f8320ff8a51ffa1 Mon Sep 17 00:00:00 2001 From: jaeyeopme Date: Sun, 7 Jun 2026 00:02:13 +0900 Subject: [PATCH] feat(rules): map and validate job export rows Add pure mapping, validation, and duplicate-handling rules for job export rows. Invalid rows are rejected before duplicate handling, and the first valid job_id wins. Tested: uv run pytest tests/test_mapping.py tests/test_validation.py tests/test_duplicate_handling.py -v Tested: uv run ruff check src/clean_data_export_api/mapping.py src/clean_data_export_api/validation.py tests Tested: uv run basedpyright Tested: uv run pytest tests/test_mapping.py tests/test_validation.py tests/test_duplicate_handling.py --cov=clean_data_export_api.mapping --cov=clean_data_export_api.validation --cov-report=term-missing Tested: uv run pytest -v Tested: uv run pytest --cov=clean_data_export_api --cov-report=term-missing --cov-fail-under=90 Tested: uv run python -m compileall src tests --- src/clean_data_export_api/mapping.py | 76 ++++++++++++++ src/clean_data_export_api/validation.py | 132 ++++++++++++++++++++++++ tests/test_duplicate_handling.py | 93 +++++++++++++++++ tests/test_mapping.py | 95 +++++++++++++++++ tests/test_validation.py | 98 ++++++++++++++++++ 5 files changed, 494 insertions(+) create mode 100644 src/clean_data_export_api/mapping.py create mode 100644 src/clean_data_export_api/validation.py create mode 100644 tests/test_duplicate_handling.py create mode 100644 tests/test_mapping.py create mode 100644 tests/test_validation.py diff --git a/src/clean_data_export_api/mapping.py b/src/clean_data_export_api/mapping.py new file mode 100644 index 0000000..b55ebac --- /dev/null +++ b/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")) diff --git a/src/clean_data_export_api/validation.py b/src/clean_data_export_api/validation.py new file mode 100644 index 0000000..39ea171 --- /dev/null +++ b/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 diff --git a/tests/test_duplicate_handling.py b/tests/test_duplicate_handling.py new file mode 100644 index 0000000..5809788 --- /dev/null +++ b/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}, + ) diff --git a/tests/test_mapping.py b/tests/test_mapping.py new file mode 100644 index 0000000..03d3012 --- /dev/null +++ b/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, + ) diff --git a/tests/test_validation.py b/tests/test_validation.py new file mode 100644 index 0000000..d702cc2 --- /dev/null +++ b/tests/test_validation.py @@ -0,0 +1,98 @@ +from datetime import UTC, datetime +from typing import TypedDict + +import pytest + +from clean_data_export_api.mapping import JobExportCandidate +from clean_data_export_api.models import RejectedRow +from clean_data_export_api.validation import ( + ValidatedJob, + partition_valid_jobs, + validate_job_candidate, +) + + +class CandidateOverrides(TypedDict, total=False): + job_id: str | None + customer_name: str | None + job_status: str | None + work_type: str | None + scheduled_date: str | None + + +def test_validate_job_candidate_accepts_valid_candidate() -> None: + result = validate_job_candidate(_candidate()) + + assert isinstance(result, ValidatedJob) + assert result.source_row_id == "row-001" + assert result.clean_job.job_id == "JOB-1001" + assert result.clean_job.customer_name == "Avery Plumbing" + assert result.clean_job.job_status == "scheduled" + assert result.clean_job.work_type == "repair" + assert result.clean_job.scheduled_date.isoformat() == "2026-06-02" + assert result.raw_record["source_row_id"] == "row-001" + + +@pytest.mark.parametrize( + ("candidate_kwargs", "reason_code"), + [ + ({"job_id": None}, "missing_job_id"), + ({"job_id": ""}, "missing_job_id"), + ({"customer_name": None}, "missing_customer_name"), + ({"customer_name": ""}, "missing_customer_name"), + ({"work_type": None}, "missing_work_type"), + ({"work_type": ""}, "missing_work_type"), + ({"job_status": "waiting"}, "invalid_status"), + ({"scheduled_date": None}, "invalid_date"), + ({"scheduled_date": "2026-15-01"}, "invalid_date"), + ], +) +def test_validate_job_candidate_rejects_invalid_candidate( + candidate_kwargs: CandidateOverrides, + reason_code: str, +) -> None: + candidate = _candidate(**candidate_kwargs) + + result = validate_job_candidate(candidate) + + assert isinstance(result, RejectedRow) + assert result.reason_code == reason_code + assert result.source_row_id == candidate.source_row_id + assert result.source_job_id == candidate.job_id + assert result.raw_record == candidate.raw_record + + +def test_partition_valid_jobs_preserves_order_and_rejections() -> None: + valid_candidate = _candidate(source_row_id="row-001", job_id="JOB-1001") + invalid_candidate = _candidate(source_row_id="row-002", job_id=None) + another_valid_candidate = _candidate(source_row_id="row-003", job_id="JOB-1003") + + valid_jobs, rejected_rows = partition_valid_jobs( + [valid_candidate, invalid_candidate, another_valid_candidate] + ) + + assert [job.source_row_id for job in valid_jobs] == ["row-001", "row-003"] + assert [row.source_row_id for row in rejected_rows] == ["row-002"] + assert rejected_rows[0].reason_code == "missing_job_id" + + +def _candidate( + *, + source_row_id: str = "row-001", + job_id: str | None = "JOB-1001", + 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}, + )