You run an ingestion job Monday. The same files get queued again Wednesday because a failed upload triggered a retry. Or two different teams independently drop the same document into your pipeline. SHA-256 doesn't care which scenario it is. The same bytes always produce the same hash. That hash is your natural idempotency key.
This article builds a Python worker that uses that property to deduplicate anchoring requests before they hit the API. It also handles the case where the API already knows about a hash your local state doesn't.
Two Dedup Layers, Not One
There are two places to catch duplicates.
Layer 1: local state. A SQLite database tracking every hash you've successfully anchored. Before calling the API, check whether the hash is already in your state file. If it is, skip the call entirely. Zero API usage, zero quota consumption.
Layer 2: the API's duplicate_of field. When you submit a hash that's already anchored, the API returns the new proof record plus a duplicate_of field pointing to the original proof ID. This catches what local state misses: two workers running in parallel, a state file that was deleted and rebuilt, or a client who anchored the same file independently before you did.
Both layers matter. Local state prevents wasted API calls. The API check is the safety net for everything else.
The State Layer
SQLite handles this well. JSON files don't survive concurrent writes safely.
import sqlite3
from pathlib import Path
DB_PATH = Path("anchor_state.db")
def init_db(conn: sqlite3.Connection) -> None:
conn.execute("""
CREATE TABLE IF NOT EXISTS anchored (
sha256 TEXT PRIMARY KEY,
proof_id TEXT NOT NULL,
anchored_at TEXT
)
""")
conn.commit()
def already_anchored(conn: sqlite3.Connection, sha256: str) -> dict | None:
row = conn.execute(
"SELECT proof_id, anchored_at FROM anchored WHERE sha256 = ?",
(sha256,)
).fetchone()
if row:
return {"proof_id": row[0], "anchored_at": row[1]}
return None
def record_anchor(
conn: sqlite3.Connection,
sha256: str,
proof_id: str,
anchored_at: str,
) -> None:
conn.execute(
"INSERT OR IGNORE INTO anchored (sha256, proof_id, anchored_at) VALUES (?, ?, ?)",
(sha256, proof_id, anchored_at),
)
conn.commit()INSERT OR IGNORE is the key detail. If two threads race to insert the same sha256, one wins and the other silently discards. No exception, no duplicate row, no corrupted state.
The already_anchored function runs before every API call. If it returns a result, the worker logs the existing proof ID and moves on.
Hashing and the API Call
Large files need chunked reading. Loading gigabytes into memory at once fails in production.
import hashlib
import requests
import time
def hash_file(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()
def call_anchor_api(sha256: str, filename: str, api_key: str) -> dict:
while True:
resp = requests.post(
"https://proofledger.io/api/v1/proof",
headers={"Authorization": f"Bearer {api_key}"},
json={"sha256": sha256, "filename": filename},
timeout=30,
)
if resp.status_code == 429:
retry_after = int(resp.headers.get("Retry-After", "60"))
print(f"Rate limited. Waiting {retry_after}s.")
time.sleep(retry_after)
continue
resp.raise_for_status()
return resp.json()The 429 retry loop reads the Retry-After header and waits. Don't blindly retry on 400, 401, or 403. Those are caller errors: bad hash format, invalid key, or wrong plan tier. raise_for_status() converts them to exceptions. The caller decides what to do with them.
Handling duplicate_of
from pathlib import Path
def process_file(
filepath: str,
api_key: str,
conn: sqlite3.Connection,
) -> dict:
sha256 = hash_file(filepath)
filename = Path(filepath).name
# Layer 1: local check
existing = already_anchored(conn, sha256)
if existing:
return {
"file": filename,
"sha256": sha256,
"status": "skipped",
"proof_id": existing["proof_id"],
}
# Layer 2: API call
try:
result = call_anchor_api(sha256, filename, api_key)
except requests.exceptions.HTTPError as e:
return {"file": filename, "sha256": sha256, "status": "error", "error": str(e)}
proof_id = result.get("id", "")
anchored_at = result.get("created_at", "")
duplicate_of = result.get("duplicate_of")
if duplicate_of:
# The canonical anchor predates this one. Record the original.
record_anchor(conn, sha256, duplicate_of, anchored_at)
return {
"file": filename,
"sha256": sha256,
"status": "duplicate",
"canonical_proof_id": duplicate_of,
}
record_anchor(conn, sha256, proof_id, anchored_at)
return {
"file": filename,
"sha256": sha256,
"status": "anchored",
"proof_id": proof_id,
}When duplicate_of is present, store the original proof ID, not the new one. If someone later asks "what's the proof for this file?", the answer is the earliest anchor. That's the one with the timestamp that matters for a pre-loss documentation argument.
Wiring the Worker
import os
import sqlite3
from pathlib import Path
API_KEY = os.environ.get("PROOFLEDGER_API_KEY", "sk_YOUR_KEY_HERE")
INCOMING = Path("./incoming")
def run(directory: Path, api_key: str) -> None:
conn = sqlite3.connect(str(DB_PATH))
init_db(conn)
files = [f for f in directory.iterdir() if f.is_file()]
print(f"Processing {len(files)} files.")
counts: dict[str, int] = {"anchored": 0, "skipped": 0, "duplicate": 0, "error": 0}
for filepath in files:
result = process_file(str(filepath), api_key, conn)
status = result["status"]
counts[status] = counts.get(status, 0) + 1
ref = (
result.get("proof_id")
or result.get("canonical_proof_id")
or result.get("error", "")
)
print(f" [{status}] {result['file']} -> {ref}")
conn.close()
print(
f"\nAnchored: {counts['anchored']}, Skipped: {counts['skipped']}, "
f"Duplicate: {counts['duplicate']}, Errors: {counts['error']}"
)
if __name__ == "__main__":
run(INCOMING, API_KEY)Run this against the same directory twice. The second run should show all files as skipped with zero API calls. That's the behavior you want.
Verifying a File After Anchoring
Once a file is anchored, the proof lives at https://proofledger.io/cert/<proof_id>. Verification doesn't need the API. The verify-proof package does it entirely offline:
from verify_proof import hash_file as vp_hash_file, verify_proof, load_proof
# proof.json downloaded from the certificate page or stored by your pipeline
proof = load_proof("evidence_photo_proof.json")
file_hash = vp_hash_file("evidence_photo.jpg")
result = verify_proof(file_hash, proof)
if result["verified"]:
print(f"Verified on {result['blockchain']} at {result['anchored_at']}")
else:
print(f"Not verified: {result['error']}")No network connection required. Useful for forensic environments with restricted internet access, or for opposing counsel who wants to verify independently without hitting your infrastructure.
Note that verify_proof checks hash integrity and Merkle path structure. It does not make a live blockchain call to confirm the tx_id exists on-chain. For that, use the explorer_url from the proof record or GET /api/v1/verify?hash=<sha256> (public, no auth, 120 requests/hr per IP).
A Few Production Notes
Concurrent workers. SQLite serializes writes but allows concurrent reads. INSERT OR IGNORE handles the race correctly when two workers hash the same file at the same time. For high-volume pipelines with many parallel workers, move the state layer to Postgres. The queries above don't change, only the connection string does.
State file loss. If anchor_state.db gets deleted, the worker falls through to Layer 2. The API catches any hashes it's already seen and returns duplicate_of. You'll make more API calls than necessary on that first recovery run, but no data is lost and no incorrect anchors get created.
Monthly quota. The API call still happens in the duplicate_of case, and it still counts against your monthly proof quota. Local state is what prevents that consumption on repeated runs. Keep the state file around, and back it up alongside your proof records.
---
The underlying idea is that SHA-256 is a pure function: same input, same output, every time. No coordination protocol needed. No distributed lock. The hash is the key, and the key is stable across retries, restarts, and concurrent workers.
Source for the verify-proof package is at github.com/Fulcrum-Enterprises/verify-proof.
What are you using for state tracking in your file ingestion pipelines? SQLite, Postgres, Redis? Curious what other approaches people are running in production.