81 lines
2.1 KiB
Python
81 lines
2.1 KiB
Python
import hashlib
|
|
import json
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
from django.db import transaction
|
|
|
|
from commerce.models import SourceRecord, SourceSystem
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class IngestResult:
|
|
record: SourceRecord
|
|
created: bool
|
|
changed: bool
|
|
|
|
|
|
def payload_hash(payload: dict[str, Any]) -> str:
|
|
canonical = json.dumps(
|
|
payload,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
default=str,
|
|
)
|
|
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
|
|
|
|
|
|
@transaction.atomic
|
|
def ingest_source_record(
|
|
*,
|
|
source: SourceSystem,
|
|
entity_type: str,
|
|
external_id: str,
|
|
payload: dict[str, Any],
|
|
source_updated_at: datetime | None = None,
|
|
) -> IngestResult:
|
|
"""Idempotently retain the latest payload and report whether it changed."""
|
|
|
|
digest = payload_hash(payload)
|
|
existing = (
|
|
SourceRecord.objects.select_for_update()
|
|
.filter(
|
|
source=source,
|
|
entity_type=entity_type,
|
|
external_id=str(external_id),
|
|
)
|
|
.first()
|
|
)
|
|
if existing is not None:
|
|
changed = (
|
|
existing.payload_hash != digest
|
|
or existing.source_updated_at != source_updated_at
|
|
)
|
|
if changed:
|
|
existing.payload = payload
|
|
existing.payload_hash = digest
|
|
existing.source_updated_at = source_updated_at
|
|
existing.save(
|
|
update_fields=[
|
|
"payload",
|
|
"payload_hash",
|
|
"source_updated_at",
|
|
"received_at",
|
|
]
|
|
)
|
|
else:
|
|
existing.save(update_fields=["received_at"])
|
|
return IngestResult(record=existing, created=False, changed=changed)
|
|
|
|
record = SourceRecord.objects.create(
|
|
source=source,
|
|
entity_type=entity_type,
|
|
external_id=str(external_id),
|
|
payload=payload,
|
|
payload_hash=digest,
|
|
source_updated_at=source_updated_at,
|
|
)
|
|
return IngestResult(record=record, created=True, changed=True)
|