You set up an evidence file intake. Files arrive throughout the day: reports, photos, PDFs from field adjusters. Some are duplicates. The same photo emailed and then uploaded through the portal. A retry loop that fires twice because the first request timed out. A processor copying from two different source folders.
If you anchor every incoming file without deduplicating first, you end up with multiple proof records for the same content. That's not a catastrophic failure. But the earlier anchor is the one that matters. It establishes when the file existed. Every anchor after the first is noise, and if your pipeline can't tell the difference, you can't reliably cite the earliest timestamp when it counts.
SHA-256 solves this cleanly. Two identical files always produce the same hash. The hash is the natural idempotency key.
Two Tiers, One Rule
The strategy here has two stages.
Local SQLite first. Before touching any API, hash the incoming file and check a local database. If that digest is already in the table, return the existing proof record and skip the network call. Fast, offline, free.
API dedup signal second. If the same hash was anchored from a different client, or before this local DB existed, POST /api/v1/proof returns a duplicate_of field pointing to the original proof ID. That original is the canonical anchor.
Both tiers enforce the same rule: the earliest anchor wins.
Setting Up Local State
pip install requests
The SQLite table is straightforward. SHA-256 is the primary key.
import sqlite3
def init_db(db_path: str = "anchors.db") -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS anchors (
sha256 TEXT PRIMARY KEY,
proof_id TEXT NOT NULL,
anchored_at TEXT,
status TEXT,
certificate_url TEXT
)
""")
conn.commit()
return conn
Inserts use INSERT OR IGNORE. If the hash is already present, the insert is a no-op and the first record stays.
def store_anchor(
conn: sqlite3.Connection,
sha256: str,
proof_id: str,
anchored_at: str,
status: str,
certificate_url: str,
) -> None:
conn.execute(
"""
INSERT OR IGNORE INTO anchors
(sha256, proof_id, anchored_at, status, certificate_url)
VALUES (?, ?, ?, ?, ?)
""",
(sha256, proof_id, anchored_at, status, certificate_url),
)
conn.commit()
No upsert. No conflict resolution. First record wins.
Hashing Files
Chunk the read so large files don't blow up memory. A 4GB video and a 10KB text file take the same path.
import hashlib
def compute_sha256(filepath: str) -> str:
h = hashlib.sha256()
with open(filepath, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
This returns the hex digest you'll send to the API. The file itself goes nowhere.
Calling the API
The v1 proof endpoint takes a SHA-256 hex digest, an optional filename, and an optional bitcoin_requested flag. Auth is a Bearer token from your ProofLedger account.
import time
import requests
from typing import Optional
API_URL = "https://proofledger.io/api/v1/proof"
API_KEY = "sk_YOUR_KEY_HERE"
def call_anchor_api(digest: str, filename: Optional[str] = None) -> dict:
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
body: dict = {"sha256": digest}
if filename:
body["filename"] = filename
for attempt in range(3):
resp = requests.post(API_URL, json=body, headers=headers, timeout=30)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", 60))
time.sleep(retry_after)
continue
if resp.status_code in (400, 401, 403):
resp.raise_for_status()
resp.raise_for_status()
return resp.json()
raise RuntimeError(f"Anchor API failed after 3 attempts for digest {digest}")
A few things worth calling out.
Rate-limit responses (429) parse the Retry-After header before sleeping. Don't use a fixed interval. The API tells you exactly how long to wait, and that value shifts based on your tier's reset window.
400, 401, and 403 are non-retryable. A malformed hash, a bad key, or a permissions mismatch won't fix itself on retry. Raise immediately and let the caller decide.
Timeouts and 5xx responses fall through the loop naturally.
The Dedup Worker
Here's where both tiers connect.
import os
def get_cached_anchor(conn: sqlite3.Connection, sha256: str) -> Optional[dict]:
row = conn.execute(
"SELECT proof_id, anchored_at, status, certificate_url FROM anchors WHERE sha256 = ?",
(sha256,),
).fetchone()
if row:
return {
"proof_id": row[0],
"anchored_at": row[1],
"status": row[2],
"certificate_url": row[3],
}
return None
def anchor_file(filepath: str, conn: sqlite3.Connection) -> dict:
digest = compute_sha256(filepath)
filename = os.path.basename(filepath)
# Tier 1: local cache
cached = get_cached_anchor(conn, digest)
if cached:
return {"source": "local_cache", "duplicate": True, "sha256": digest, **cached}
# Tier 2: API
data = call_anchor_api(digest, filename)
# If duplicate_of is present, the API has seen this hash before.
# Use the original proof ID, not the new record's ID.
canonical_id = data.get("duplicate_of") or data["id"]
store_anchor(
conn,
sha256=digest,
proof_id=canonical_id,
anchored_at=data.get("created_at", ""),
status=data.get("status", "pending"),
certificate_url=data.get("certificate_url", ""),
)
return {
"source": "api",
"duplicate": "duplicate_of" in data,
"sha256": digest,
"proof_id": canonical_id,
"status": data.get("status"),
"certificate_url": data.get("certificate_url"),
}
The duplicate_of field is the API's signal that it's already seen this hash. It contains the ID of the original proof record. Store that ID, not the new record's ID.
This distinction matters for evidence documentation. If a document was anchored three months ago and now arrives in a new intake batch, the relevant timestamp is the one from three months ago. That anchor establishes the file existed before whatever event is now in dispute. Citing last Tuesday's duplicate record instead is technically valid but gives the wrong timestamp.
Running It
if __name__ == "__main__":
import sys
conn = init_db()
if len(sys.argv) < 2:
print("Usage: python worker.py <file> [file ...]")
sys.exit(1)
for path in sys.argv[1:]:
result = anchor_file(path, conn)
print(
f"{path}: "
f"source={result['source']}, "
f"duplicate={result['duplicate']}, "
f"proof_id={result['proof_id']}"
)
Run it twice on the same file:
$ python worker.py inspection-report.pdf
inspection-report.pdf: source=api, duplicate=False, proof_id=prf_abc123
$ python worker.py inspection-report.pdf
inspection-report.pdf: source=local_cache, duplicate=True, proof_id=prf_abc123
Second run never touches the API. Same proof ID both times.
Extending From Here
A few natural additions depending on your pipeline.
Status polling. The initial anchor returns status: "pending". Poll GET /api/v1/proof/:id on a background schedule and update the status column when it flips to anchored. The proof ID is already in your table, so this is just a lookup and an update.
Third-party verification. Once anchored, GET /api/v1/verify?hash=<sha256> is public with no API key required. Pass the hex digest and the endpoint returns the proof record and blockchain explorer links. An auditor or opposing party can check the anchor without touching your credentials.
File-watcher integration. This worker composes cleanly with a watchdog or chokidar watcher. The watcher fires on new files, calls anchor_file(), and the SQLite table handles deduplication automatically. You don't need separate filename or modification-time tracking. The hash covers it.
Multi-worker state. SQLite works for a single-node worker. If you're running multiple workers against the same intake folder, you'll want a shared state store. PostgreSQL with INSERT ... ON CONFLICT DO NOTHING gives you the same semantics. The canonical_id = data.get("duplicate_of") or data["id"] logic stays unchanged.
For offline verification of anchored files, the verify-proof PyPI package handles local Merkle proof validation with no network call.
What do you use for idempotency state in file ingestion pipelines? SQLite, Redis, Postgres? Curious how others have handled the multi-worker case without a central DB.
