You set up a file intake. Files arrive throughout the day: reports, photos, PDFs from several sources. Some are duplicates. The same photo emailed and then uploaded through a web form. 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 object carrying the original proof's id and created_at. 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        INTEGER 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: int,
    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 external_id (your own record id; resending it returns the same proof with HTTP 200 and already_existed: true). 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 is the time left in the current one-minute window.

400, 401, and 403 are non-retryable. A malformed hash, a bad key, or a disabled account 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.
    # It is an object: use its id, not the new record's id.
    dup = data.get("duplicate_of")
    canonical_id = dup["id"] if dup else data["id"]

    store_anchor(
        conn,
        sha256=digest,
        proof_id=canonical_id,
        anchored_at=data.get("polygon_anchored_at") or "",
        status=data.get("polygon_status", "PENDING"),
        certificate_url=data.get("certificate_url", ""),
    )

    return {
        "source": "api",
        "duplicate": dup is not None,
        "sha256": digest,
        "proof_id": canonical_id,
        "status": data.get("polygon_status"),
        "certificate_url": data.get("certificate_url"),
    }

The duplicate_of field is the API's signal that it has seen this hash before, from any account. It is an object with the id and created_at of the original proof record. Store that ID, not the new record's ID. Your own record, data["id"], is still created, and it is the one GET /api/v1/proof/{id} and the bundle endpoint serve, since those are owner-only; keep it as well if you will need the bundle.

This distinction matters when the date is the point. 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 is what shows the file existed by then. Citing last Tuesday's duplicate record instead is technically valid but gives the later date.

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=42

$ python worker.py inspection-report.pdf
inspection-report.pdf: source=local_cache, duplicate=True, proof_id=42

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 response carries polygon_status: "PENDING". Poll GET /api/v1/proof/{id} on a background schedule and update the status column when it flips to ANCHORED, usually within a minute; polygon_anchored_at and chain_tx fill in at the same time. 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. Anyone holding the file 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 = dup["id"] if dup else data["id"] logic stays unchanged.

For offline verification of anchored files, the verify-proof PyPI package checks a file against its saved proof bundle (GET /api/v1/proof/{id}/bundle) with no network call, and the bundle's Polygon transaction can be looked up on any block explorer. The API reference is at https://proofledger.io/api.html.

---

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.