A12: Add meaningful changes/provenance to ImportRecordEvent, skip events for unchanged data
This commit is contained in:
@@ -189,12 +189,30 @@ def _import_records_locked(session: Session, *, url: str, region: str, category:
|
|||||||
if report is None:
|
if report is None:
|
||||||
report = CatchReport(fish=fish, waterbody=waterbody, bait=bait, spot=None, weight_g=raw.weight_g, caught_at=caught, reported_at=now, player_name=raw.player, source_type=SourceType.official_record, source_url=url, source_external_id=key, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload=payload)
|
report = CatchReport(fish=fish, waterbody=waterbody, bait=bait, spot=None, weight_g=raw.weight_g, caught_at=caught, reported_at=now, player_name=raw.player, source_type=SourceType.official_record, source_url=url, source_external_id=key, source_confidence=100, moderation_status=ModerationStatus.approved, raw_payload=payload)
|
||||||
session.add(report)
|
session.add(report)
|
||||||
session.add(ImportRecordEvent(catch_report=report, import_run=run, event_type="created", created_at=now))
|
session.add(ImportRecordEvent(
|
||||||
|
catch_report=report, import_run=run, event_type="created", created_at=now,
|
||||||
|
changes={"weight_g": raw.weight_g, "player": raw.player, "record_date": raw.record_date.isoformat()},
|
||||||
|
provenance={"source_system": "rf4-official", "source_url": url, "source_external_id": key},
|
||||||
|
))
|
||||||
run.rows_created += 1
|
run.rows_created += 1
|
||||||
else:
|
else:
|
||||||
|
# A12: Only create event if actual values changed
|
||||||
|
old_payload = (report.raw_payload or {})
|
||||||
|
new_payload = asdict(raw) | {"record_date": raw.record_date.isoformat()}
|
||||||
|
changed_fields = {}
|
||||||
|
for field in ("weight_g", "player", "waterbody", "bait", "record_date"):
|
||||||
|
old_val = old_payload.get(field)
|
||||||
|
new_val = new_payload.get(field)
|
||||||
|
if old_val != new_val:
|
||||||
|
changed_fields[field] = {"old": old_val, "new": new_val}
|
||||||
|
if changed_fields:
|
||||||
report.raw_payload = payload
|
report.raw_payload = payload
|
||||||
report.source_url = url
|
report.source_url = url
|
||||||
session.add(ImportRecordEvent(catch_report=report, import_run=run, event_type="updated", created_at=now))
|
session.add(ImportRecordEvent(
|
||||||
|
catch_report=report, import_run=run, event_type="updated", created_at=now,
|
||||||
|
changes=changed_fields,
|
||||||
|
provenance={"source_system": "rf4-official", "source_url": url, "source_external_id": key},
|
||||||
|
))
|
||||||
run.rows_updated += 1
|
run.rows_updated += 1
|
||||||
run.status = ImportStatus.success
|
run.status = ImportStatus.success
|
||||||
run.finished_at = datetime.now(timezone.utc)
|
run.finished_at = datetime.now(timezone.utc)
|
||||||
|
|||||||
@@ -216,12 +216,14 @@ class ExternalEntityAlias(Base):
|
|||||||
|
|
||||||
|
|
||||||
class ImportRecordEvent(Base):
|
class ImportRecordEvent(Base):
|
||||||
"""Track per-record import events for D09 revision history."""
|
"""Track per-record import events for D09 revision history with provenance."""
|
||||||
__tablename__ = "import_record_event"
|
__tablename__ = "import_record_event"
|
||||||
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
|
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
|
||||||
catch_report_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("catch_report.id"), index=True)
|
catch_report_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("catch_report.id"), index=True)
|
||||||
import_run_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("official_record_import.id"), index=True)
|
import_run_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("official_record_import.id"), index=True)
|
||||||
event_type: Mapped[str] = mapped_column(String(20)) # created/updated/deleted
|
event_type: Mapped[str] = mapped_column(String(20)) # created/updated/deleted
|
||||||
|
changes: Mapped[dict | None] = mapped_column(JSON, default=None) # what fields changed
|
||||||
|
provenance: Mapped[dict | None] = mapped_column(JSON, default=None) # source system, external_id
|
||||||
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
|
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
|
||||||
catch_report: Mapped[CatchReport] = relationship()
|
catch_report: Mapped[CatchReport] = relationship()
|
||||||
import_run: Mapped[OfficialRecordImport] = relationship()
|
import_run: Mapped[OfficialRecordImport] = relationship()
|
||||||
|
|||||||
@@ -47,7 +47,8 @@ def test_parser_and_import_are_idempotent() -> None:
|
|||||||
first = import_records(db, url="fixture://records", region="RU", category="records", html=html)
|
first = import_records(db, url="fixture://records", region="RU", category="records", html=html)
|
||||||
second = import_records(db, url="fixture://records", region="RU", category="records", html=html)
|
second = import_records(db, url="fixture://records", region="RU", category="records", html=html)
|
||||||
assert (first.rows_created, first.rows_updated) == (2, 0)
|
assert (first.rows_created, first.rows_updated) == (2, 0)
|
||||||
assert (second.rows_created, second.rows_updated) == (0, 2)
|
# A12: Second import of identical data creates no events (no fields changed)
|
||||||
|
assert (second.rows_created, second.rows_updated) == (0, 0)
|
||||||
assert db.scalar(select(func.count()).select_from(CatchReport).where(CatchReport.source_type == SourceType.official_record)) == 2
|
assert db.scalar(select(func.count()).select_from(CatchReport).where(CatchReport.source_type == SourceType.official_record)) == 2
|
||||||
assert db.scalar(select(func.count()).select_from(OfficialRecordImport)) == 2
|
assert db.scalar(select(func.count()).select_from(OfficialRecordImport)) == 2
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user