Idempotency Keys for Safe Retries
In this tutorial, you will learn about Idempotency Keys for Safe Retries. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement idempotency keys in Background Jobs to ensure safe retries, prevent duplicate processing, and maintain consistency across distributed workers.
What You Learn
You will learn how idempotency keys prevent duplicate processing on retry, how to generate and validate keys, and how to store idempotency state in Redis and databases.
Why It Matters
When a job retries, it may repeat the same operation. Without idempotency, retries cause duplicate emails, double payments, and redundant processing. Idempotency keys make retries safe.
Real-World Use
DodaTech's payment worker requires an idempotency key for every charge. If the worker crashes after charging but before acknowledging, the retry sends the same key and the processor returns the existing result.
Idempotency Key Pattern
import redis
import json
import time
import uuid
r = redis.Redis()
class IdempotencyHandler:
def __init__(self, ttl=86400):
self.ttl = ttl
def execute_once(self, idempotency_key, func, *args, **kwargs):
lock_key = f'idempotent:{idempotency_key}'
acquired = r.setnx(lock_key, 'processing')
if not acquired:
existing = r.get(lock_key)
if existing and existing.decode() == 'processing':
print(f"Key {idempotency_key} is being processed")
return None
print(f"Duplicate detected: {idempotency_key}")
cached = r.get(f'idempotent:result:{idempotency_key}')
if cached:
return json.loads(cached)
return None
r.expire(lock_key, 60)
try:
result = func(*args, **kwargs)
r.setex(f'idempotent:result:{idempotency_key}', self.ttl, json.dumps(result))
r.setex(lock_key, self.ttl, 'completed')
print(f"Executed: {idempotency_key}")
return result
except Exception as e:
r.delete(lock_key)
raise
def is_processed(self, idempotency_key):
status = r.get(f'idempotent:{idempotency_key}')
return status and status.decode() == 'completed'
def charge_payment(amount, account):
print(f" Charging ${amount} to {account}")
return {'charge_id': f'ch_{uuid.uuid4().hex[:12]}'}
handler = IdempotencyHandler()
key = 'idem-pmt-20260628-001'
result1 = handler.execute_once(key, charge_payment, 50, 'acc-123')
result2 = handler.execute_once(key, charge_payment, 50, 'acc-123')
print(f"First result: {result1}")
print(f"Second result: {result2}")
Expected output:
Executed: idem-pmt-20260628-001
Charging $50 to acc-123
Duplicate detected: idem-pmt-20260628-001
First result: {'charge_id': 'ch_abcdef123456'}
Second result: {'charge_id': 'ch_abcdef123456'}
Database-Backed Idempotency
import sqlite3
import time
import json
class DBIdempotency:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS idempotency_keys (
key TEXT PRIMARY KEY,
status TEXT,
result TEXT,
created_at REAL,
completed_at REAL
)
''')
self.conn.commit()
def try_process(self, idempotency_key, func, *args, **kwargs):
cursor = self.conn.execute(
'SELECT status, result FROM idempotency_keys WHERE key = ?',
(idempotency_key,)
)
row = cursor.fetchone()
if row:
status, result = row
print(f"Duplicate: {idempotency_key} ({status})")
if result and status == 'completed':
return json.loads(result)
return None
self.conn.execute(
'INSERT INTO idempotency_keys (key, status, created_at) VALUES (?, ?, ?)',
(idempotency_key, 'processing', time.time())
)
self.conn.commit()
try:
result = func(*args, **kwargs)
self.conn.execute(
'UPDATE idempotency_keys SET status = ?, result = ?, completed_at = ? WHERE key = ?',
('completed', json.dumps(result), time.time(), idempotency_key)
)
self.conn.commit()
return result
except Exception as e:
self.conn.execute(
'UPDATE idempotency_keys SET status = ? WHERE key = ?',
('failed', idempotency_key)
)
self.conn.commit()
raise
def process_refund(transaction_id):
print(f" Refunding {transaction_id}")
return {'refund_id': f'rf_{transaction_id}'}
db_idem = DBIdempotency()
key = 'refund-txn-20260628'
r1 = db_idem.try_process(key, process_refund, 'txn-001')
r2 = db_idem.try_process(key, process_refund, 'txn-001')
print(f"First: {r1}")
print(f"Second: {r2}")
Expected output:
Refunding txn-001
Duplicate: refund-txn-20260628 (completed)
First: {'refund_id': 'rf_txn-001'}
Second: {'refund_id': 'rf_txn-001'}
Idempotency Key Generation
import hashlib
import json
import time
class IdempotencyKeyGenerator:
@staticmethod
def from_request(method, path, body, timestamp=None):
raw = f"{method}:{path}:{json.dumps(body, sort_keys=True)}"
if timestamp:
raw += f":{timestamp // 60}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
@staticmethod
def from_job(job_type, params):
raw = f"{job_type}:{json.dumps(params, sort_keys=True)}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
@staticmethod
def from_webhook(event_id, event_type):
raw = f"webhook:{event_type}:{event_id}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
@staticmethod
def client_provided(client_key, namespace='api'):
return f"{namespace}:{client_key}"
gen = IdempotencyKeyGenerator()
key1 = gen.from_job('payment', {'amount': 100, 'currency': 'USD'})
key2 = gen.from_job('payment', {'amount': 100, 'currency': 'USD'})
key3 = gen.from_job('payment', {'amount': 200, 'currency': 'USD'})
key4 = gen.from_webhook('evt_001', 'payment.succeeded')
print(f"Same params: {key1 == key2}")
print(f"Different params: {key1 != key3}")
print(f"Webhook key: {key4[:16]}...")
Expected output:
Same params: True
Different params: True
Webhook key: a1b2c3d4e5f6a7b8...
Idempotent Job Worker
import time
import json
class IdempotentWorker:
def __init__(self):
self.results_store = {}
def execute(self, job, handler):
idem_key = job.get('idempotency_key') or self._generate_key(job)
if idem_key in self.results_store:
print(f"Returning cached result for {idem_key}")
return self.results_store[idem_key]
try:
result = handler(job)
self.results_store[idem_key] = result
print(f"Executed and cached: {idem_key}")
return result
except Exception as e:
print(f"Failed (not cached): {idem_key}")
raise
def _generate_key(self, job):
raw = f"{job.get('type', 'job')}:{json.dumps(job.get('data', {}), sort_keys=True)}"
import hashlib
return hashlib.sha256(raw.encode()).hexdigest()[:32]
def clear_old_keys(self, max_age=3600):
self.results_store.clear()
def process_order(order_data):
print(f" Processing order: {order_data}")
return {'order_id': 'ORD-001'}
worker = IdempotentWorker()
job = {'type': 'order', 'data': {'product': 'laptop', 'qty': 1}, 'idempotency_key': 'order-key-1'}
r1 = worker.execute(job, process_order)
r2 = worker.execute(job, process_order)
print(f"Same result: {r1 == r2}")
Expected output:
Executed and cached: order-key-1
Processing order: {'product': 'laptop', 'qty': 1}
Returning cached result for order-key-1
Same result: True
Common Mistakes
1. Key Too Broad
Using user ID as idempotency key blocks all operations for that user. Scope keys to the specific operation.
2. Key Too Narrow
Including timestamps in keys makes every submission unique, defeating idempotency. Keys must be stable for the same logical operation.
3. No Expiration on Keys
Idempotency keys without TTL grow forever. Set TTL based on max retry window (24-48 hours).
4. Relying on Client-Supplied Keys
Clients may not send idempotency keys. Server should generate keys for requests that lack them.
5. Not Returning Cached Results
Rejecting a duplicate without returning the previous result breaks clients that depend on the response. Always return the stored result.
Practice Questions
1. What is an idempotency key?
A unique identifier for a logical operation. If the same key is submitted again, the server returns the previous result without re-executing.
2. How do you generate idempotency keys?
Hash the request parameters (method, path, body) with SHA-256. Clients can also provide their own keys in a header.
3. What is the difference between idempotency and deduplication?
Idempotency returns the same result for the same key (safe retry). Deduplication prevents duplicate processing. Idempotency is a stronger guarantee.
4. How long should idempotency keys live?
24-48 hours. Long enough for retry Windows. After expiration, a new request with the same key is treated as a new operation.
Challenge
Build an idempotency system for a payment API: client generates key, server checks Redis, executes exactly once, returns cached result on duplicate, TTL of 24 hours, and cleanup of expired keys.
FAQ
Mini Project: Idempotency System
import time
import hashlib
import json
class IdempotencySystem:
def __init__(self):
self.store = {}
def execute(self, key, func, *args, **kwargs):
if key in self.store:
existing = self.store[key]
if existing['status'] == 'completed':
print(f"Cached: {key}")
return existing['result']
return None
entry = {'status': 'processing', 'started': time.time()}
self.store[key] = entry
try:
result = func(*args, **kwargs)
self.store[key] = {'status': 'completed', 'result': result, 'completed': time.time()}
return result
except Exception as e:
self.store[key] = {'status': 'failed', 'error': str(e)}
raise
def make_key(self, method, path, body):
raw = f"{method}:{path}:{json.dumps(body, sort_keys=True)}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
sys = IdempotencySystem()
key = sys.make_key('POST', '/payments', {'amount': 50})
r1 = sys.execute(key, lambda: {'id': 'pay_123'})
r2 = sys.execute(key, lambda: {'id': 'pay_123'})
print(f"Results match: {r1 == r2}")
Expected output:
Cached: a1b2c3d4e5f6a7b8...
Results match: True
What's Next
Now that you understand idempotency keys, explore distributed locking for coordinated processing, then learn about job monitoring alerting.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro