Use an idempotency key generated before processing. Store the result in a table with the key as a unique constraint. On retry, attempt to insert the result; if a duplicate key error occurs, fetch and return the existing result.
def process_job(job_id, idempotency_key):
try:
result = do_work(job_id)
db.execute(
"INSERT INTO job_results (idempotency_key, result) VALUES (?, ?)",
idempotency_key, result
)
except UniqueViolation:
result = db.query(
"SELECT result FROM job_results WHERE idempotency_key = ?",
idempotency_key
)
return result
Consumer
idempotency_key = generate_uuid()
try:
result = process_job(job.id, idempotency_key)
queue.ack(message)
except Exception:
queue.nack(message) # retry with same key
The first successful write commits the result. Subsequent retries detect the existing key and reuse the original output, preventing duplicate side effects while preserving exactly-once semantics.