from datetime import datetime
from app.extensions import db
class IdempotencyKey(db.Model):
__tablename__ = "idempotency_keys"
__table_args__ = (
db.UniqueConstraint("key", "endpoint", name="uq_idem_key_endpoint"),
)
id = db.Column(db.BigInteger, primary_key=True)
key = db.Column(db.String(255), nullable=False)
endpoint = db.Column(db.String(255), nullable=False)
request_hash = db.Column(db.String(64), nullable=False)
status = db.Column(db.String(20), nullable=False, default="in_progress")
response_code = db.Column(db.Integer, nullable=True)
response_body = db.Column(db.Text, nullable=True)
created_at = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
completed_at = db.Column(db.DateTime, nullable=True)
def mark_completed(self, code, body):
self.status = "completed"
self.response_code = code
self.response_body = body
self.completed_at = datetime.utcnow()import hashlib
import json
from functools import wraps
from flask import request, jsonify
from sqlalchemy.exc import IntegrityError
from app.extensions import db
from app.models.idempotency import IdempotencyKey
def _hash_body():
raw = request.get_data(cache=True) or b""
return hashlib.sha256(raw).hexdigest()
def idempotent(view):
@wraps(view)
def wrapper(*args, **kwargs):
key = request.headers.get("Idempotency-Key")
if not key:
return jsonify(error="Idempotency-Key header required"), 400
endpoint = request.endpoint
req_hash = _hash_body()
record = IdempotencyKey(key=key, endpoint=endpoint, request_hash=req_hash)
db.session.add(record)
try:
db.session.flush() # forces the unique-constraint check now
except IntegrityError:
db.session.rollback()
return _replay(key, endpoint, req_hash)
# We own the key: run the real handler exactly once.
rv = view(*args, **kwargs)
response = _coerce(rv)
record.mark_completed(response.status_code, response.get_data(as_text=True))
db.session.commit()
return response
return wrapper
def _replay(key, endpoint, req_hash):
existing = (
IdempotencyKey.query
.filter_by(key=key, endpoint=endpoint)
.one()
)
if existing.request_hash != req_hash:
return jsonify(error="Idempotency-Key reused with different payload"), 409
if existing.status != "completed":
return jsonify(error="Original request still in progress"), 409
return existing.response_body, existing.response_code, {"Content-Type": "application/json"}
def _coerce(rv):
from flask import make_response
return make_response(rv)from flask import Blueprint, request, jsonify
from app.extensions import db
from app.models.order import Order
from app.services.payments import charge_card
from app.decorators.idempotency import idempotent
bp = Blueprint("orders", __name__, url_prefix="/api/orders")
@bp.post("")
@idempotent
def create_order():
payload = request.get_json(silent=True) or {}
amount = payload.get("amount_cents")
if not amount or amount <= 0:
return jsonify(error="amount_cents must be positive"), 422
order = Order(
customer_id=payload["customer_id"],
amount_cents=amount,
status="pending",
)
db.session.add(order)
db.session.flush() # get order.id before charging
charge = charge_card(customer_id=order.customer_id, amount_cents=amount)
order.status = "paid"
order.charge_id = charge.id
return jsonify(id=order.id, status=order.status, charge_id=order.charge_id), 201This snippet shows how a Flask API can safely deduplicate POST requests using an Idempotency-Key header, so a client that retries after a timeout does not create two orders or charge a card twice. The pattern is essential for any non-idempotent endpoint that clients may retry: the server records the outcome of the first request keyed by the client-supplied token and replays that stored response for every subsequent request carrying the same key.
The IdempotencyKey model tab defines the durable store. Each row holds the key, the endpoint and request_hash it was first used against, a status (in_progress or completed), and the serialized response_code/response_body once the work finishes. A unique constraint on (key, endpoint) is the linchpin: it lets the database, not the application, enforce that only one row per key can ever be inserted, which is what makes the check race-free under concurrent retries. Storing request_hash also allows the code to detect a key being reused with a different body, a common client bug that should fail loudly rather than silently replay the wrong response.
The idempotent decorator tab wraps a view. It reads the header, hashes the JSON body, and attempts an INSERT inside a transaction. If the insert succeeds, this is the first request and the wrapped handler runs; its response is captured and written back to the row as completed. If the insert raises IntegrityError, another request already owns the key: the code loads the existing row and either replays the stored response, reports a 409 if the same key arrives with a mismatched request_hash, or returns 409 while the original is still in_progress. That in-progress state is the tricky edge case — it protects against two near-simultaneous retries both executing the body.
The orders blueprint tab shows the realistic usage: a create_order view decorated with @idempotent, doing genuine side-effecting work. The trade-offs are worth noting — keys need a retention/expiry policy, and the response is only replayed verbatim, so non-deterministic fields are frozen at first execution. Reach for this whenever at-least-once delivery meets exactly-once semantics.
Related snips
timestamp = request.headers.fetch('X-Signature-Timestamp')
signature = request.headers.fetch('X-Signature')
payload = request.raw_post
data = "#{timestamp}.#{payload}"
expected = OpenSSL::HMAC.hexdigest('SHA256', ENV.fetch('WEBHOOK_SECRET'), data)HMAC signed API requests for webhook and partner integrity
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
Share this code
Here's the card — post it anywhere.
