Skip to content
AITroveRead. Build. Understand.
Reading list
Make this comfortable

Python SQLite job project: reject conflicting replays by request identity

Last updated: 30 Sept 20264 min read
tutorial
IntermediateBy AITrove Editorial

An idempotent request key associates one operation identity with one accepted payload so a matching replay can reuse its stored result.

Download Python source kit

Operation contract

The local job ledger accepts a bounded job ID and an exact nonnegative integer amount. It stores the accepted payload and result together under a unique key. The first call creates the row; a matching replay returns the same result without another row. Reusing the key for a different amount raises a conflict instead of silently returning an unrelated result.

Failure and ownership boundary

This fixture calculates a local result inside one SQLite transaction and uses one connection sequentially. It does not dispatch an external payment, resolve ambiguous remote timeouts or coordinate competing connections. Those cases need backend-specific contention handling and a durable delivery design. Django atomic transactions: rollback the batch and defer callbacks and Python unittest.mock: inject a collaborator failure before publication explain why replay identity alone cannot promise exactly-once external effects.

Working program

python
import re
import sqlite3

def open_job_ledger():
    database = sqlite3.connect(":memory:")
    database.execute("CREATE TABLE jobs (job_id TEXT PRIMARY KEY, amount INTEGER NOT NULL, result INTEGER NOT NULL)")
    return database

def accept_job(database, job_id, amount):
    if not isinstance(job_id, str) or re.fullmatch(r"J-[0-9]{4}", job_id) is None:
        raise ValueError("job identity rejected")
    if type(amount) is not int or not 0 <= amount <= 1000000:
        raise ValueError("amount rejected")
    with database:
        existing = database.execute("SELECT amount, result FROM jobs WHERE job_id = ?", (job_id,)).fetchone()
        if existing is not None:
            if existing[0] != amount:
                raise ValueError("request identity conflict")
            return existing[1], False
        result = amount + 25
        database.execute("INSERT INTO jobs VALUES (?, ?, ?)", (job_id, amount, result))
        return result, True

if __name__ == "__main__":
    database = open_job_ledger()
    try:
        print(accept_job(database, "J-0041", 125))
        print(accept_job(database, "J-0041", 125))
        try:
            accept_job(database, "J-0041", 250)
        except ValueError:
            print("conflicting replay rejected")
        print("stored jobs:", database.execute("SELECT count(*) FROM jobs").fetchone()[0])
    finally:
        database.close()

Output

Output
(150, True)
(150, False)
conflicting replay rejected
stored jobs: 1

Costs and limits

The primary-key index supports keyed retrieval with database-dependent logarithmic work. The fixture retains one row per accepted key and has no expiry policy. Deleting an identity too early can permit a later replay to be treated as a new operation.

Common Mistakes

  • A matching key with a different payload must be rejected.
  • A local transaction does not make an external side effect exactly once.

Connected lessons

Django atomic transactions: rollback the batch and defer callbacks, Python SQLite project: transactional batches, duplicate IDs and reopen checks, Python unittest.mock: inject a collaborator failure before publication.

Check the next state boundary

Python SQLite outbox project: commit a receipt and its pending event together.

Follow the service contract

Python SQLite outbox leases: stale workers must not acknowledge a newer claim, Python HMAC envelopes: authenticate exact bytes and separate replay policy.

Follow the ownership and update boundary

Python SQLite lease renewal: reject stale tokens and exact-expiry ownership.

Check this related boundary

Python exercise: retain first-seen order while rejecting duplicate IDs.

python
idempotent-jobs
Storage details