Idempotency in APIs: Preventing Duplicate Operations

Idempotent:
  Set price = $100         → Run 3 times → price = $100 ✓
  DELETE /orders/123       → Run 3 times → order deleted once ✓
  GET /users/123           → Run 100 times → same user data ✓
 
Not idempotent:
  Increment counter by 1   → Run 3 times → counter +3, not +1 ✗
  POST /orders (new order) → Run 3 times → 3 orders created ✗
  Charge card $100         → Run 3 times → $300 charged ✗

Client → POST /payments
         Idempotency-Key: uuid-abc-123
         Body: { amount: 100, card: ... }
 
Server:
  1. Check if uuid-abc-123 was seen before
  2. If yes → return stored response (don't process again)
  3. If no → process payment, store response against uuid-abc-123

import stripe
 
stripe.api_key = "sk_..."
 
# First call — processes payment
result = stripe.PaymentIntent.create(
    amount=1000,
    currency="usd",
    idempotency_key="order_123_payment_attempt_1"
)
 
# Retry (network failed, client retries) — returns same result, no duplicate charge
result = stripe.PaymentIntent.create(
    amount=1000,
    currency="usd",
    idempotency_key="order_123_payment_attempt_1"  # Same key
)

import redis
import json
import uuid
from fastapi import FastAPI, Header, HTTPException, Request
from datetime import timedelta
 
app = FastAPI()
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
 
IDEMPOTENCY_TTL = timedelta(days=1)  # Store keys for 24 hours
 
def idempotent(ttl: timedelta = IDEMPOTENCY_TTL):
    def decorator(func):
        async def wrapper(*args, idempotency_key: str = Header(None), **kwargs):
            if not idempotency_key:
                raise HTTPException(status_code=400, detail="Idempotency-Key header required")
            
            cache_key = f"idem:{idempotency_key}"
            
            # Check if we've seen this key before
            cached = r.get(cache_key)
            if cached:
                # Return the stored response — don't process again
                return json.loads(cached)
            
            # Process the request
            result = await func(*args, **kwargs)
            
            # Store the result
            r.setex(cache_key, int(ttl.total_seconds()), json.dumps(result))
            
            return result
        return wrapper
    return decorator
 
@app.post("/payments")
@idempotent()
async def create_payment(request: Request, idempotency_key: str = Header(None)):
    body = await request.json()
    
    # Process payment (only runs once per idempotency key)
    payment_id = process_payment(body["amount"], body["card_token"])
    
    return {"payment_id": payment_id, "status": "success"}

# Database-backed approach (more durable than Redis alone)
from sqlalchemy import Column, String, JSON, DateTime
from datetime import datetime, timedelta
 
class IdempotencyRecord(Base):
    __tablename__ = "idempotency_records"
    
    key = Column(String, primary_key=True)
    response = Column(JSON)
    created_at = Column(DateTime, default=datetime.utcnow)
    expires_at = Column(DateTime)
 
def get_or_create_idempotent_response(key: str, handler):
    record = db.query(IdempotencyRecord).filter_by(key=key).first()
    
    if record and record.expires_at > datetime.utcnow():
        return record.response  # Return cached response
    
    # Not seen before — process and store
    result = handler()
    
    record = IdempotencyRecord(
        key=key,
        response=result,
        expires_at=datetime.utcnow() + timedelta(days=1)
    )
    db.add(record)
    db.commit()
    
    return result

import redis.lock
 
def process_idempotent(key: str, handler):
    cache_key = f"idem:{key}"
    lock_key = f"idem_lock:{key}"
    
    # Try to get cached response first (fast path)
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Acquire lock to prevent concurrent processing
    with r.lock(lock_key, timeout=30):
        # Check again inside lock (another request may have processed while we waited)
        cached = r.get(cache_key)
        if cached:
            return json.loads(cached)
        
        # Process once
        result = handler()
        r.setex(cache_key, 86400, json.dumps(result))
        return result

import uuid
 
# Simple: UUID per user action
key = str(uuid.uuid4())
 
# Or deterministic: hash of operation + user ID (dangerous if parameters change)
import hashlib
key = hashlib.sha256(f"{user_id}:{order_id}:{amount}:{timestamp_minute}".encode()).hexdigest()