DEV Community: Muhammad Umair Virk

Idempotency Keys for Python Payment APIs

Muhammad Umair Virk — Sat, 29 Aug 2026 17:31:43 +0000

A user double-taps Pay. The network retries. Your API creates two orders for one charge.

POST is not idempotent. If you handle payments without an idempotency strategy, you will eventually ship a bug that finance notices before engineering does.

I've debugged checkout flows in production — post-payment 500s, guest checkout edge cases, gateway parity. Here are the patterns I use for idempotent payment endpoints in Python.

Companion on Medium: What Production Checkout Taught Me About Idempotency — reconciliation, webhooks, and state machines.

1. The problem in one request

POST /orders HTTP/1.1
Content-Type: application/json

{"amount": 9900, "currency": "AED", "payment_method_id": "pm_abc"}

The client sends this. The server charges the card and creates an order. The response never reaches the app. The client retries the same POST.

Without protection: two orders, or one order and one orphan charge.

2. Idempotency keys — the standard fix

The client generates a unique key once per user action and sends it on every retry:

POST /orders HTTP/1.1
Content-Type: application/json
Idempotency-Key: 7c9e6679-7425-40de-944b-e07fc1f90ae7

{"amount": 9900, "currency": "AED", "payment_method_id": "pm_abc"}

Server rules:

First request with key K → process normally, store (K → response)
Duplicate request with key K → return stored response, do not re-charge

Stripe popularized this header. You can implement the same pattern on any checkout API.

3. Database schema

CREATE TABLE idempotency_keys (
    key         VARCHAR(64) PRIMARY KEY,
    request_hash TEXT NOT NULL,
    response_body JSONB NOT NULL,
    status_code  INT NOT NULL,
    created_at   TIMESTAMPTZ DEFAULT NOW()
);

-- Optional: tie to business entity
CREATE UNIQUE INDEX orders_idempotency_key_idx ON orders (idempotency_key);

Store the key before calling the payment gateway when possible (with status processing) to win race conditions between concurrent retries.

4. FastAPI implementation

from fastapi import FastAPI, Header, HTTPException, Request
from sqlalchemy.orm import Session
import hashlib
import json

app = FastAPI()

def request_hash(body: dict) -> str:
    return hashlib.sha256(json.dumps(body, sort_keys=True).encode()).hexdigest()


@app.post("/orders")
async def create_order(
    body: dict,
    db: Session = Depends(get_db),
    idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"),
):
    if not idempotency_key:
        raise HTTPException(400, "Idempotency-Key header required")

    existing = db.query(IdempotencyRecord).filter_by(key=idempotency_key).first()
    if existing:
        if existing.request_hash != request_hash(body):
            raise HTTPException(422, "Idempotency key reused with different body")
        return JSONResponse(existing.response_body, status_code=existing.status_code)

    # Reserve the key (status: processing) — prevents duplicate charges under concurrency
    record = IdempotencyRecord(key=idempotency_key, request_hash=request_hash(body), status="processing")
    db.add(record)
    db.commit()

    try:
        charge = payment_gateway.charge(body)
        order = Order(amount=body["amount"], charge_id=charge.id, idempotency_key=idempotency_key)
        db.add(order)
        response = {"order_id": order.id, "status": "paid"}
        record.response_body = response
        record.status_code = 201
        record.status = "completed"
        db.commit()
        return JSONResponse(response, status_code=201)
    except Exception:
        db.rollback()
        raise

Key details:

Reject keys reused with a different body (422)
Store response before returning to client
Use a DB transaction around order + idempotency record

5. Flask — lighter version

from flask import Flask, request, jsonify
from functools import wraps

app = Flask(__name__)

# Production: Redis or Postgres, not in-memory
_idempotency_store = {}


@app.post("/orders")
def create_order():
    key = request.headers.get("Idempotency-Key")
    if not key:
        return jsonify({"error": "Idempotency-Key required"}), 400

    body = request.get_json()
    if key in _idempotency_store:
        stored = _idempotency_store[key]
        if stored["body_hash"] != hash(body):
            return jsonify({"error": "Key reused with different payload"}), 422
        return jsonify(stored["response"]), stored["status"]

    charge = charge_customer(body)
    order_id = save_order(body, charge.id)
    response = {"order_id": order_id, "status": "paid"}
    _idempotency_store[key] = {"body_hash": hash(body), "response": response, "status": 201}
    return jsonify(response), 201

Use Redis with TTL (24–72 hours) or Postgres in production — not a module-level dict.

6. Gateway-level idempotency

Most gateways accept their own idempotency key:

# Stripe example
stripe.PaymentIntent.create(
    amount=9900,
    currency="aed",
    idempotency_key=idempotency_key,  # same key as your API
)

Defense in depth: idempotency at your API layer and at the gateway. Your order service should not depend on the gateway alone — webhooks and retries can still duplicate rows.

7. Status reconciliation

Sometimes the charge succeeds but your API crashes before saving the order. Run a reconciliation job:

def reconcile_orphan_charges():
    pending = db.query(Order).filter_by(status="payment_pending").all()
    for order in pending:
        gateway_status = payment_gateway.get_charge(order.charge_id)
        if gateway_status == "succeeded":
            order.status = "paid"
        elif gateway_status == "failed":
            order.status = "failed"
    db.commit()

Webhooks (payment_intent.succeeded) are the other half — never rely only on the synchronous response.

8. Guest checkout edge case

Guest orders often have user_id: null in JSON. Some serializers treat missing field and null differently. Downstream services that expect "no user" as absent field break when they receive explicit null.

Normalize at the API boundary:

def normalize_guest(body: dict) -> dict:
    if body.get("user_id") is None:
        body.pop("user_id", None)  # or store as NULL consistently — pick one
    return body

Parity bugs here cause post-payment 500s that look like payment failures but are mapping issues.

Checklist

[ ] Idempotency-Key required on POST /orders and POST /payments
[ ] Store key → response in DB or Redis with TTL
[ ] Reject same key + different body (422)
[ ] Pass idempotency key through to payment gateway
[ ] Webhook handler updates order status
[ ] Reconciliation job for stuck payment_pending orders
[ ] Unique constraint on orders.idempotency_key

Docker Compose for Python Backends — A Minimal Production Stack

Muhammad Umair Virk — Sat, 29 Aug 2026 17:29:34 +0000

Most Python backends eventually leave "run on my laptop." The first container deploy is where teams discover that localhost is not a universal address and that a Dockerfile without a health check is a black box to the load balancer.

This is the minimal stack I use and review in PRs: FastAPI, a hardened Dockerfile, Compose for local parity with production topology.

Companion article on Medium: Why Serverless Engineers Already Understand Containers — same patterns, more on mental models.

1. Dockerfile essentials

FROM python:3.12-slim

WORKDIR /app

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

RUN useradd --create-home appuser
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app ./app

USER appuser
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')"

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Review checklist:

[ ] Non-root user — container escape should not mean root on the host
[ ] Slim base — smaller attack surface and faster pulls
[ ] HEALTHCHECK hits a real endpoint that verifies dependencies if possible
[ ] No secrets in layers — use env injection at runtime

2. docker-compose.yml

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      ENVIRONMENT: local-compose

docker compose up --build
curl http://localhost:8000/health

If /health returns a hostname that changes between restarts, you are talking to the container — not a stale local process.

3. The rule that prevents the most outages

Inside Compose, localhost is the container itself.

When you add Postgres:

services:
  api:
    build: .
    environment:
      DATABASE_URL: postgresql://app:secret@db:5432/appdb
    depends_on:
      - db
  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_USER: app
      POSTGRES_PASSWORD: secret
      POSTGRES_DB: appdb

The API must use host db, not 127.0.0.1. This single mistake causes more first-week container incidents than any Dockerfile syntax error.

4. Health endpoint

import os
import socket
from fastapi import APIRouter

router = APIRouter()

@router.get("/health")
def health():
    return {
        "status": "healthy",
        "hostname": socket.gethostname(),
        "environment": os.getenv("ENVIRONMENT", "development"),
    }

Orchestrators, load balancers, and Compose use this to decide whether to route traffic. A listening socket that cannot serve requests should fail the check.

5. Local vs. production gaps

Concern	Local Compose	Production
Secrets	`.env` (gitignored)	Secrets Manager / platform env
Database	Named volume	Managed RDS + migrations
Logs	stdout	Structured JSON → aggregation
Deploy	Manual	CI pipeline → registry → roll out

Closing these gaps deliberately — not on the day of launch — is what separates a demo from a service.

AWS Secrets Manager Patterns for Python Microservices

Muhammad Umair Virk — Sat, 29 Aug 2026 17:29:33 +0000

Hardcoding secrets is the fastest way to get paged at 3 AM — or worse, flagged in a security audit.

Centralized secret management is not optional once you run more than one microservice. I have implemented shared retrieval layers in Python (Flask, FastAPI, Lambda) using AWS Secrets Manager. These are the patterns that survive production — caching, least privilege, rotation, and what breaks when you skip any of them.

Pattern 1: Never bake secrets into images

# ❌ NEVER
DATABASE_URL = "postgresql://user:password123@host/db"

# ✅ Runtime injection
import os
DATABASE_URL = os.environ["DATABASE_URL"]

In Docker, use .env files locally and AWS-injected env vars in production. Add .env to .dockerignore and .gitignore.

Pattern 2: Centralized retrieval helper

import boto3
import json
from functools import lru_cache

client = boto3.client("secretsmanager", region_name="me-south-1")


@lru_cache(maxsize=32)
def get_secret(secret_name: str) -> dict:
    response = client.get_secret_value(SecretId=secret_name)
    return json.loads(response["SecretString"])


def get_database_url() -> str:
    secrets = get_secret("prod/taskflow/database")
    return (
        f"postgresql://{secrets['username']}:{secrets['password']}"
        f"@{secrets['host']}:{secrets['port']}/{secrets['dbname']}"
    )

Why @lru_cache? Secrets Manager charges per API call. Cache in-process for Lambda warm instances and long-running containers. Invalidate on rotation events if needed.

Pattern 3: IAM least privilege

Each service gets its own IAM role with access to only its secrets:

{
  "Effect": "Allow",
  "Action": "secretsmanager:GetSecretValue",
  "Resource": "arn:aws:secretsmanager:me-south-1:123456789:secret:prod/taskflow/*"
}

Never use one mega-role for all secrets across all services.

Pattern 4: Lambda vs. long-running services

Runtime	Strategy
Lambda	Fetch on cold start, cache in module scope
ECS / Docker	Fetch on startup, cache in app singleton
Local dev	`.env` file or LocalStack

# Lambda — module-level cache survives warm invocations
_db_url = None

def get_db_url():
    global _db_url
    if _db_url is None:
        _db_url = get_database_url()
    return _db_url

Pattern 5: Secret rotation

Enable automatic rotation in Secrets Manager for database credentials. Your app must handle connection drops gracefully:

from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool

engine = create_engine(get_database_url(), pool_pre_ping=True)

pool_pre_ping=True verifies connections before use — critical after rotation.

Pattern 6: Structured logging — never log secrets

import logging
logger = logging.getLogger(__name__)

# ❌ logger.info(f"Connecting to {DATABASE_URL}")
# ✅
logger.info("database_connection_initialized", extra={"host": secrets["host"]})

Redact secrets in error handlers and APM tools.

Pattern 7: Compose locally, Secrets Manager in prod

# docker-compose.yml (local)
services:
  api:
    env_file:
      - .env

# ECS task definition (prod)
secrets:
  - name: DATABASE_URL
    valueFrom: arn:aws:secretsmanager:...:secret:prod/taskflow/database-url

Same app code. Different secret source per environment.

What I built

At work, I implemented a shared secret retrieval layer used across Python and Go services — standardizing how services access database credentials, API keys, and third-party tokens. Result: zero hardcoded secrets in repos, audit-friendly access logs, and rotation without redeploying application code.

Checklist for your next project

[ ] All secrets in Secrets Manager (or Parameter Store for non-sensitive config)
[ ] .env in .gitignore and .dockerignore
[ ] IAM role per service, least privilege
[ ] In-process caching with rotation awareness
[ ] pool_pre_ping on database connections
[ ] No secrets in logs, CI output, or error messages

DEV Community: Muhammad Umair Virk

Idempotency Keys for Python Payment APIs

1. The problem in one request

2. Idempotency keys — the standard fix

3. Database schema

4. FastAPI implementation

5. Flask — lighter version

6. Gateway-level idempotency

7. Status reconciliation

8. Guest checkout edge case

Checklist

Further reading

Docker Compose for Python Backends — A Minimal Production Stack

1. Dockerfile essentials

2. docker-compose.yml

3. The rule that prevents the most outages

4. Health endpoint

5. Local vs. production gaps

Further reading

AWS Secrets Manager Patterns for Python Microservices

Pattern 1: Never bake secrets into images

Pattern 2: Centralized retrieval helper

Pattern 3: IAM least privilege

Pattern 4: Lambda vs. long-running services

Pattern 5: Secret rotation

Pattern 6: Structured logging — never log secrets

Pattern 7: Compose locally, Secrets Manager in prod

What I built

Checklist for your next project

Further reading