Deduplicate AI workflow requests with Python and SQLite

A retry should not create a second job

A customer submits a brief, the connection times out, and the browser tries again. If each attempt immediately starts an AI summary or creates a CRM record, a single request can become duplicate work. An idempotency key identifies one intended operation across those attempts. The application must remember the key; simply attaching it to a request changes nothing.

This AI-assisted editorial tutorial builds a local request ledger, not a complete queue. The Python example was tested with synthetic data on Python 3.13.15 and SQLite 3.50.4. Tests covered repeated submissions, changed content, tenant separation, eight concurrent attempts and recovery after a database lock. No AI API, CRM or customer account was contacted. Official documentation was checked October 4, 2026.

1. Decide what counts as the same request

Our example uses the pair tenant plus request_key as its identity. The caller keeps the same key when retrying the same operation. A genuinely new operation gets a new key, even if its text happens to match. In a real application, derive the tenant from authenticated server context; never let a public form choose another customer’s namespace.

The stored fingerprint covers a small, validated payload with service and message fields. Sorting JSON keys makes dictionary order irrelevant. Whitespace inside a message remains meaningful, so editing it produces a conflict. Decide your own normalization rules before generating the fingerprint and keep them stable across retries. The hash is a comparison aid, not encryption or authorization.

SQLite can enforce uniqueness on a composite primary key. ON CONFLICT with DO NOTHING leaves an existing row intact rather than replacing its contents. This syntax requires SQLite 3.24.0 or later. Here, the conflict target is explicit: tenant and request_key. [1]

2. Save this request ledger as ledger_demo.py

Use Python 3.12 or newer with the sqlite3 module available. This example explicitly selects autocommit=True and then controls the transaction with SQL BEGIN, COMMIT and ROLLBACK. Python’s commit() and rollback() methods have no effect in that mode. Values are bound through question-mark placeholders. [2]

BEGIN IMMEDIATE starts a write transaction before inserting or inspecting the row. Another writer may make it fail with SQLITE_BUSY; Python exposes database failures as exceptions. The connection timeout here is two seconds. Keep this transaction short and do not put an AI call inside it. [2] [3]

from contextlib import closing
import hashlib
import json
import sqlite3


def initialize(path):
    with closing(sqlite3.connect(path, autocommit=True)) as db:
        db.execute("""CREATE TABLE IF NOT EXISTS requests (
            tenant TEXT NOT NULL,
            request_key TEXT NOT NULL,
            payload_hash TEXT NOT NULL,
            payload_json TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'pending',
            PRIMARY KEY (tenant, request_key)
        )""")


def register(path, tenant, key, payload):
    if not all(isinstance(v, str) and 1 <= len(v) <= 100
               for v in (tenant, key)):
        raise ValueError('invalid tenant or key')
    if not isinstance(payload, dict) or set(payload) != {'service', 'message'}:
        raise ValueError('expected service and message')
    if not all(isinstance(v, str) and 1 <= len(v) <= 4000
               for v in payload.values()):
        raise ValueError('invalid payload values')
    body = json.dumps(payload, sort_keys=True, separators=(',', ':'),
                      ensure_ascii=False, allow_nan=False)
    digest = hashlib.sha256(body.encode('utf-8')).hexdigest()
    db = sqlite3.connect(path, timeout=2.0, autocommit=True)
    try:
        db.execute('BEGIN IMMEDIATE')
        inserted = db.execute('''
            INSERT INTO requests (tenant, request_key, payload_hash, payload_json)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(tenant, request_key) DO NOTHING
        ''', (tenant, key, digest, body)).rowcount == 1
        saved_hash, status = db.execute('''
            SELECT payload_hash, status FROM requests
            WHERE tenant = ? AND request_key = ?
        ''', (tenant, key)).fetchone()
        if saved_hash != digest:
            raise ValueError('key reused with different content')
        db.execute('COMMIT')
        return {'created': inserted, 'status': status}
    except Exception:
        if db.in_transaction:
            db.execute('ROLLBACK')
        raise
    finally:
        db.close()

3. Append a disposable demonstration and run it

Append the following block to the same file, then run python ledger_demo.py in a terminal. The temporary database contains only the fictional payload and is removed when the block finishes. No package installation or API key is needed.

The first printed dictionary should have created set to True and status pending. The second should have created set to False with the same status. The final line should say key reused with different content. Those are the observed outcomes of the local example, not a claim that a downstream job executed.

from pathlib import Path
from tempfile import TemporaryDirectory

with TemporaryDirectory() as folder:
    path = Path(folder) / 'requests.sqlite'
    initialize(path)
    payload = {'service': 'automation', 'message': 'Review a fictional inquiry.'}
    print(register(path, 'demo-team', 'request-001', payload))
    print(register(path, 'demo-team', 'request-001', payload))
    try:
        register(path, 'demo-team', 'request-001',
                 {**payload, 'message': 'A different request.'})
    except ValueError as error:
        print(error)

4. Test persistence, contention and isolation

For an application, initialize its database during setup and use a configured path on durable local storage. The temporary directory above is deliberately unsuitable for production persistence. Opening a new connection for each registration lets the demonstration exercise the saved record rather than a Python dictionary in memory.

Before connecting a worker, repeat these checks with a disposable database. A lock timeout is a failure to register work; do not catch it and return a success receipt. A controlled retry must reuse the original key and payload.

• Send eight concurrent calls with the same tenant, key and payload: exactly one should report created=True and one row should remain.
• Reuse the key with changed content: reject it while preserving the original stored payload.
• Use the same key under a second trusted tenant: allow an independent record.
• Hold a write transaction open in another connection: registration should time out; release the lock and confirm a later retry succeeds.
• Reopen the database and inspect the stored rows. Test the actual backup and restore process separately before real intake.

5. Treat pending as a record, not a completed action

The example stops after recording pending work. It does not claim jobs, run a model, send messages or mark completion. created=False means this request was already registered; it does not mean processing finished. A production API should return a receipt and an honest status based on its stored record.

A separate worker needs atomic job claiming, attempt tracking and a recovery policy. Consider the hard failure: a CRM write succeeds, then the worker crashes before saving its completion state. Retrying the external action can still create a duplicate. Use the destination’s documented idempotency support when available, or reconcile against a stable external identifier before retrying. This local ledger alone cannot promise exactly-once effects.

Keep the original input separate from AI-generated suggestions. Scope downstream permissions and retain human approval for actions that require it. Decide how long to retain keys: deleting the ledger entry also removes its duplicate protection.

6. Choose storage that matches the deployment

SQLite permits one writer at a time per database file. Its guidance recommends considering a client/server database for many simultaneous writers or multiple servers, and warns about direct access over network filesystems. [4]

Use this tutorial as a single-host exercise. If application instances have separate local disks, they will have separate ledgers and can each accept the same key. Choose shared durable storage with an equivalent uniqueness constraint for that architecture. The useful pattern is the identity, transaction and recovery contract—not copying a local database file into every deployment.

The payload is stored as readable JSON. Restrict database access, define retention and avoid real customer information during development. Add authentication, public-request validation, size limits and monitoring at the application boundary; this function is not a public endpoint.

Sources & further reading

1. SQLite: UPSERT and explicit conflict targets
2. Python: sqlite3 transaction control and parameter binding
3. SQLite: transactions and BEGIN IMMEDIATE
4. SQLite: appropriate uses and concurrency limits

Make your AI workflows safe to retry

Sociologix can help design intake records, job status, review steps and recovery paths across your website, AI workers and business systems. Bring one workflow where duplicate or uncertain results slow your team down.