External Data Tracking

ExternalDataTracker records a hashed snapshot of every external fetch a decision relied on — an API response, a database query, a file — so the decision stays reproducible even after that source changes underneath you.

external For: governance & audit

How it works

Snapshot the fetch. track_api_call hashes the response and stores it when the policy allows.
Detect drift. detect_drift compares the live value against the latest snapshot to see whether the source moved.
Append a correction. When the upstream value is fixed, correct_snapshot records the fix without erasing the original.

flowchart LR
    A["pricing_service fetch"] --> B["track_api_call()"]
    B --> C["snapshot + SHA-256 hash"]
    C --> D["classify_ticket decision"]
    E["live value, two weeks later"] --> F{"detect_drift"}
    C --> F
    F -- "changed" --> G["correct_snapshot<br/>original kept"]

Install

pip install briefcase-ai[external]

from briefcase.external import (
    ExternalDataTracker,
    SnapshotPolicy,
    SnapshotFrequency,
)

Track an External Call

track_api_call() hashes the response, stores a snapshot when the policy allows, and reports whether the source drifted since the last snapshot.

from briefcase.external import ExternalDataTracker

tracker = ExternalDataTracker()

response = {"items": [{"sku": "A-100", "price": 19.99}]}

result = tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data=response,
    record_count=1,
)

print(result["data_hash"][:12])     # SHA-256 of the response
print(result["snapshot_stored"])    # True — first snapshot is always stored
print(result["drift_detected"])     # False — nothing to compare against yet

The same shape applies to database queries and file fetches:

tracker.track_db_query(
    db_system="postgres",
    db_name="catalog",
    query="SELECT sku, price FROM products",
    result_data=[{"sku": "A-100", "price": 19.99}],
    result_count=1,
    store_snapshot=True,
)

tracker.track_file_fetch(
    source_name="reference_rates",
    file_data=b"sku,price\nA-100,19.99\n",
    file_path="reference/rates.csv",
    record_count=1,
)

Method	Use it for
`track_api_call`	An HTTP/API response the decision read.
`track_db_query`	A database query result the decision read.
`track_file_fetch`	A reference file the decision read.

Detect Drift

Compare current data against the latest stored snapshot for a source. detect_drift() returns None when there is no prior snapshot.

tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data={"items": [{"sku": "A-100", "price": 19.99}]},
)

report = tracker.detect_drift(
    "pricing_service",
    current_data={"items": [{"sku": "A-100", "price": 24.99}]},
)

print(report.has_changed)    # True
print(report.drift_score)    # 1.0
print(report.size_delta)     # byte difference vs the baseline snapshot

Compare two specific snapshots by id with compare_snapshots():

tracker = ExternalDataTracker()

first = tracker.track_api_call(
    api_name="inventory_api", endpoint="https://inv.internal/v1/stock", method="GET",
    response_data={"items": [{"sku": "A-100", "qty": 40}]},
)
second = tracker.track_api_call(
    api_name="inventory_api", endpoint="https://inv.internal/v1/stock", method="GET",
    response_data={"items": [{"sku": "A-100", "qty": 25}]},
)

report = tracker.compare_snapshots(first["snapshot_id"], second["snapshot_id"])
print(report.has_changed)    # True

Snapshot Policy

A SnapshotPolicy controls when snapshots are taken and how long they are kept. Set a per-source policy, or pass default_policy to the tracker.

from briefcase.external import (
    ExternalDataTracker, SnapshotPolicy, SnapshotFrequency,
)

tracker = ExternalDataTracker(
    default_policy=SnapshotPolicy(
        frequency=SnapshotFrequency.ON_CHANGE,
        retention_days=90,
        max_snapshots=100,
    )
)

tracker.set_policy(
    "pricing_service",
    SnapshotPolicy(frequency=SnapshotFrequency.EVERY_CALL),
)

Field	Default	Description
`frequency`	`ON_CHANGE`	When to store a snapshot
`retention_days`	`90`	Days to retain snapshots (`0` = forever)
`change_threshold`	`0.0`	Minimum change to count as drift on `ON_CHANGE`
`max_snapshots`	`0`	Max snapshots per source (`0` = unlimited)
`compress`	`False`	Compress snapshot bodies before storage

SnapshotFrequency values: EVERY_CALL, ON_CHANGE, HOURLY, DAILY, WEEKLY.

Append a Correction

When a source returned bad data, append a correction instead of overwriting the snapshot. The correction keeps the parent’s valid_time (when the data was true in the real world) but gets a fresh transaction time, so historical queries still see the original belief and later queries see the corrected value.

original = tracker.track_api_call(
    api_name="pricing_service",
    endpoint="https://pricing.internal/v1/catalog",
    method="GET",
    response_data={"items": [{"sku": "A-100", "price": 1999.00}]},  # bad value
)

corrected = tracker.correct_snapshot(
    original["snapshot_id"],
    corrected_data={"items": [{"sku": "A-100", "price": 19.99}]},
    source="manual_review",
)

print(corrected.parent_snapshot_id == original["snapshot_id"])  # True

The original snapshot is never mutated; the correction records its lineage through parent_snapshot_id.

Redact PII Before Storage

Pass a sanitizer (for example briefcase.sanitize.Sanitizer) and snapshot bodies are redacted before they are persisted to durable storage. The data_hash is still computed over the original payload, so drift detection is unaffected.

from briefcase.external import ExternalDataTracker
from briefcase.sanitize import Sanitizer

tracker = ExternalDataTracker(sanitizer=Sanitizer())

Where this fits

Related: RAG Versioning The same snapshot-and-detect model, applied to a retrieval index.

Guide: Reproducible RAG Put snapshots and versioning together end to end.