Detecting sponsor portal record deltas with hashing

Problem statement #

You need a Python routine that compares the records a sponsor portal returns today against what you stored on the last sync, decides which ones genuinely changed, and writes only those — so that a re-run against an unchanged portal produces no database writes and no new audit-ledger rows, and a single field edit produces exactly one.

This task sits under Incremental Change Data Capture, part of the broader Automated Ingestion & Data Sync Workflows practice. The routine here is deliberately narrow: it fingerprints records, classifies each as new, changed, unchanged, or deleted, and persists only the deltas. It does not decide whether a record is valid — that boundary belongs to Schema Validation Pipelines, which each genuine delta passes through before it reaches a domain table.

Prerequisites #

• Python 3.10+ for modern type hints and datetime.now(timezone.utc).
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] for the transactional upsert. Hashing and serialization use the standard library (hashlib, json). Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • SYNC_DB_URL — the SQLAlchemy connection string for the capture-state store.
• Policy config: a version-controlled list of volatile fields to exclude from the fingerprint, kept alongside your University Policy Mapping Frameworks. Acquisition of the current portal records is handled upstream by Automating Daily Grant Portal Polling with Python requests; this page assumes today’s records are already in hand.

Step-by-step implementation #

The routine below fetches the current portal records, builds a canonical serialization of each, hashes it, compares against the stored fingerprint, classifies the change, and upserts only the deltas — appending one ledger entry per real change. The stable per-record key is what makes the whole thing safe to re-run.

Figure: each record is canonicalized and hashed once; the comparison against the stored fingerprint routes it to a skip or a single-entry upsert.

Step 1 — Fetch the current records and load stored fingerprints #

Start by pairing today’s portal records against what you already know. The stored fingerprints come from the capture-state table in one query, keyed by the portal’s native record id, so the comparison is an in-memory dictionary lookup rather than a per-record round trip.

import hashlib
import json
import logging
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session

logging.basicConfig(level=logging.INFO, format="%(asctime)s [DELTA] %(message)s")
logger = logging.getLogger(__name__)


def load_stored_hashes(session: Session, record_ids: list[str]) -> dict[str, str]:
    """One query returns the last-known fingerprint for every candidate id."""
    rows = session.execute(
        select(CaptureState.record_id, CaptureState.content_hash)
        .where(CaptureState.record_id.in_(record_ids))
    ).all()
    return {rid: h for rid, h in rows}

Step 2 — Build a canonical serialization #

This is the step that decides correctness. The fingerprint must depend only on business content, so key order is fixed, volatile server-owned fields are dropped, and every date is normalized to a UTC ISO string. Skip any of these and the hash captures serialization noise, making unchanged records look changed forever.

# Server-owned fields the portal rewrites on every export. Excluding them is
# what stops a refreshed timestamp from being mistaken for a real edit.
VOLATILE_FIELDS = frozenset({"retrieved_at", "etag", "_version", "sync_token"})


def _normalize_dates(value: Any) -> Any:
    """Coerce datetimes to a canonical UTC ISO string so tz or precision
    differences from the portal never alter the fingerprint."""
    if isinstance(value, datetime):
        return value.astimezone(timezone.utc).isoformat(timespec="seconds")
    return value


def canonical_serialization(record: dict[str, Any]) -> str:
    stable = {
        k: _normalize_dates(v)
        for k, v in record.items()
        if k not in VOLATILE_FIELDS
    }
    # sort_keys makes the string independent of the portal's key order.
    return json.dumps(stable, sort_keys=True, separators=(",", ":"), default=str)


def content_hash(record: dict[str, Any]) -> str:
    return hashlib.sha256(canonical_serialization(record).encode("utf-8")).hexdigest()

Step 3 — Classify each record #

With a stable hash and the stored fingerprints in hand, classification is a pure comparison. Handle deletes explicitly: any record id present in the stored set but absent from today’s portal pull is a tombstone candidate.

def classify_batch(
    current: dict[str, dict[str, Any]],   # record_id -> portal record
    stored: dict[str, str],               # record_id -> stored hash
) -> dict[str, list[str]]:
    """Partition record ids into new / changed / unchanged / deleted."""
    buckets: dict[str, list[str]] = {"new": [], "changed": [], "unchanged": [], "deleted": []}

    for rid, record in current.items():
        new_hash = content_hash(record)
        prior = stored.get(rid)
        if prior is None:
            buckets["new"].append(rid)
        elif prior != new_hash:
            buckets["changed"].append(rid)
        else:
            buckets["unchanged"].append(rid)

    # A record we knew about that the portal no longer returns is a soft delete.
    for rid in stored.keys() - current.keys():
        buckets["deleted"].append(rid)

    return buckets

Step 4 — Upsert only the deltas and append the ledger #

Only new, changed, and deleted records take a write. Each upsert is guarded by a content-hash comparison, and each delta appends exactly one immutable ledger entry. Unchanged records get only a refreshed last_seen_at, which proves they were observed without polluting the ledger.

from sqlalchemy.dialects.postgresql import insert as pg_insert


def apply_deltas(
    session: Session,
    current: dict[str, dict[str, Any]],
    stored: dict[str, str],
    buckets: dict[str, list[str]],
    cursor: str,
) -> dict[str, int]:
    now = datetime.now(timezone.utc)
    stats = {k: 0 for k in ("new", "changed", "unchanged", "deleted")}

    for change_type in ("new", "changed"):
        for rid in buckets[change_type]:
            new_hash = content_hash(current[rid])
            stmt = pg_insert(CaptureState).values(
                record_id=rid, source_cursor=cursor, content_hash=new_hash,
                prev_hash=stored.get(rid), last_seen_at=now, change_type=change_type,
            ).on_conflict_do_update(
                index_elements=["record_id"],
                set_={
                    "source_cursor": cursor, "prev_hash": CaptureState.content_hash,
                    "content_hash": new_hash, "last_seen_at": now,
                    "change_type": change_type,
                },
                where=(CaptureState.content_hash != new_hash),  # write only on a real change
            )
            session.execute(stmt)
            session.add(AuditLedger(
                record_id=rid, content_hash=new_hash, prev_hash=stored.get(rid),
                change_type=change_type, captured_at=now,
            ))
            stats[change_type] += 1

    for rid in buckets["deleted"]:
        session.add(AuditLedger(
            record_id=rid, content_hash=None, prev_hash=stored.get(rid),
            change_type="deleted", captured_at=now,
        ))
        stats["deleted"] += 1

    # Unchanged records: refresh last_seen_at only, no ledger entry.
    if buckets["unchanged"]:
        session.query(CaptureState).filter(
            CaptureState.record_id.in_(buckets["unchanged"])
        ).update({"last_seen_at": now}, synchronize_session=False)
        stats["unchanged"] = len(buckets["unchanged"])

    session.commit()
    logger.info("deltas applied: %s", stats)
    return stats

The where=(CaptureState.content_hash != new_hash) clause is what guarantees idempotency at the database layer: if two runs race, the second finds the hash already stored and writes nothing. A genuine delete records a tombstone ledger row with a null content_hash rather than physically removing the state row, so the change history and federal retention window stay intact.

Schema and field reference #

Verification #

1. Re-run is a no-op. Run the full routine twice against an unchanged portal pull. The second apply_deltas must report new=0, changed=0, deleted=0; only unchanged may be non-zero, and the AuditLedger row count must not move. SELECT count(*) FROM audit_ledger WHERE captured_at >= :run_start returns 0 on the second pass.
2. Reproduce a fingerprint. Re-run content_hash on a stored record’s normalized content and confirm it equals the persisted content_hash. An equal digest proves the stored fingerprint matches the record it describes.
3. Single entry per edit. Change one field of one record, run once, and confirm exactly one new ledger row exists for that record_id with change_type = 'changed' and a prev_hash equal to the record’s previous content_hash.
4. Delete detection. Remove a record from the simulated portal pull and confirm it classifies deleted with a tombstone ledger entry, while its state row is retained.

Troubleshooting #

Three gotchas specific to hash-based delta detection:

• Unstable JSON key order. json.dumps without sort_keys=True serializes keys in insertion order, so the same record re-fetched with a different key order produces a different hash and looks changed every run. Always canonicalize with sort_keys=True; if a nested object comes from the portal as a list whose order is not meaningful, sort that list before hashing too.
• Volatile fields polluting the hash. A portal that stamps retrieved_at or bumps an etag on every export will make every record hash differently even when nothing of substance changed. Keep the VOLATILE_FIELDS exclusion list in version-controlled policy config and expand it the moment a “changed” spike has no corresponding business edit — a run where 100% of records classify changed is almost always a leaked volatile field, not a real event.
• Silent deletes. Most portals drop a record rather than sending an explicit delete, so a routine that only iterates today’s pull never notices the removal. Always diff stored.keys() - current.keys() and record a tombstone; never physically delete the state row, because 2 CFR 200.333 requires the change history be retained past the award’s final report.

Frequently asked questions #

Related #

• Up to the parent topic: Incremental Change Data Capture
• Automating Daily Grant Portal Polling with Python requests — the polling layer that hands this routine today’s records
• Validating Grant Payloads with Pydantic v2 in Batch — where each genuine delta is policy-checked next
• SQLAlchemy Upsert vs Raw SQL for Idempotent Grant Records — the upsert mechanics behind the delta write