Idempotency Key em Python Puro: O Selo Que Impede Sua API de Cobrar o Cliente Duas Vezes (Mesmo Se Ele Clicar 10 Vezes)
Você já clicou duas vezes num botão de pagamento e rezou pra não ser cobrado em dobro? Pois é. Seu usuário também faz isso. E se sua API não tiver proteção contra idempotência, ela vai processar a mesma requisição duas vezes — e a culpa vai ser sua, não do usuário ansioso.
Idempotência é o conceito matemático que diz: “aplicar a mesma operação múltiplas vezes produz o mesmo resultado que aplicar uma vez.” Em APIs, isso significa que se o cliente enviar a mesma requisição com a mesma idempotency key, o servidor deve retornar o mesmo resultado — sem executar a operação novamente.
O Perrengue do Olivetto
Semana passada, um script de automação que eu mantinha processou o mesmo webhook três vezes. O motivo? Um timeout na resposta fez o cliente retryar automaticamente. Resultado: três e-mails idênticos enviados pro mesmo cliente, três registros duplicados no banco, e uma conversa constrangedora com o suporte explicando que “não, o sistema não está quebrado, é só… redundante demais.”
A solução? Implementar idempotency keys do zero, com SQLite, sem depender de Redis, sem framework, sem mágica. Só Python puro e um pouco de disciplina.
O Que é uma Idempotency Key?
É um identificador único que o cliente envia no header (geralmente Idempotency-Key ou X-Request-ID) pra sinalizar que aquela requisição é única. O servidor armazena essa key junto com o resultado da operação. Se a mesma key aparecer de novo, o servidor retorna o resultado armazenado em vez de executar tudo de novo.
Pense assim: é como um carimbo no passaporte. Você pode passar pela imigração mil vezes, mas o carimbo prova que você já entrou naquele dia.
Implementação Completa em Python Puro
Vamos construir um sistema de idempotência com três componentes:
- Storage: SQLite pra persistir keys e resultados
- Decorator: Pra aplicar idempotência em qualquer função
- Middleware HTTP: Pra interceptar requisições e verificar keys
1. Storage Layer com SQLite
import sqlite3
import json
import hashlib
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Optional, Tuple
class IdempotencyStore:
def __init__(self, db_path: str = "idempotency.db", ttl_hours: int = 24):
self.db_path = Path(db_path)
self.ttl_hours = ttl_hours
self._init_db()
def _init_db(self):
with sqlite3.connect(self.db_path) as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS idempotency_keys (
key_hash TEXT PRIMARY KEY,
original_key TEXT NOT NULL,
result_data TEXT,
status_code INTEGER,
created_at TEXT NOT NULL,
completed_at TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_created_at
ON idempotency_keys(created_at)
""")
def _hash_key(self, key: str) -> str:
return hashlib.sha256(key.encode()).hexdigest()
def get(self, key: str) -> Optional[Tuple[Any, int]]:
key_hash = self._hash_key(key)
cutoff = datetime.now() - timedelta(hours=self.ttl_hours)
with sqlite3.connect(self.db_path) as conn:
cursor = conn.execute(
"""SELECT result_data, status_code, created_at, completed_at
FROM idempotency_keys WHERE key_hash = ?""",
(key_hash,)
)
row = cursor.fetchone()
if not row:
return None
result_data, status_code, created_at, completed_at = row
created_dt = datetime.fromisoformat(created_at)
if created_dt < cutoff:
return None
if not completed_at:
return None
return json.loads(result_data), status_code
def reserve(self, key: str) -> bool:
key_hash = self._hash_key(key)
try:
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""INSERT INTO idempotency_keys
(key_hash, original_key, created_at)
VALUES (?, ?, ?)""",
(key_hash, key, datetime.now().isoformat())
)
return True
except sqlite3.IntegrityError:
return False
def complete(self, key: str, result: Any, status_code: int = 200):
key_hash = self._hash_key(key)
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"""UPDATE idempotency_keys
SET result_data = ?, status_code = ?, completed_at = ?
WHERE key_hash = ?""",
(json.dumps(result), status_code, datetime.now().isoformat(), key_hash)
)
def cleanup_expired(self):
cutoff = datetime.now() - timedelta(hours=self.ttl_hours)
with sqlite3.connect(self.db_path) as conn:
conn.execute(
"DELETE FROM idempotency_keys WHERE created_at < ?",
(cutoff.isoformat(),)
)
2. Decorator pra Funções Idempotentes
import functools
import uuid
from typing import Callable
def idempotent(store: IdempotencyStore, key_extractor: Callable[..., str] = None):
def decorator(func: Callable):
@functools.wraps(func)
def wrapper(*args, **kwargs):
if key_extractor:
idem_key = key_extractor(*args, **kwargs)
else:
idem_key = str(args[0]) if args else str(uuid.uuid4())
cached = store.get(idem_key)
if cached:
result, status = cached
return result
if not store.reserve(idem_key):
import time
for _ in range(10):
time.sleep(0.1)
cached = store.get(idem_key)
if cached:
return cached[0]
raise RuntimeError(f"Operação com key {idem_key} travada")
try:
result = func(*args, **kwargs)
store.complete(idem_key, result, 200)
return result
except Exception as e:
store.complete(idem_key, {"error": str(e)}, 500)
raise
return wrapper
return decorator
3. Uso Prático: Processamento de Pagamentos
store = IdempotencyStore("pagamentos.db", ttl_hours=48)
@idempotent(store, key_extractor=lambda txn_id, amount: txn_id)
def processar_pagamento(transaction_id: str, amount: float) -> dict:
print(f"Processando pagamento {transaction_id}: R$ {amount}")
import time
time.sleep(2)
resultado = {
"transaction_id": transaction_id,
"amount": amount,
"status": "aprovado",
"timestamp": datetime.now().isoformat()
}
return resultado
# Primeira chamada: executa normalmente
resultado1 = processar_pagamento("txn_12345", 99.90)
print(resultado1)
# Segunda chamada com MESMO transaction_id: retorna cache
resultado2 = processar_pagamento("txn_12345", 99.90)
print(resultado2)
# Chamada com transaction_id diferente: executa normalmente
resultado3 = processar_pagamento("txn_67890", 149.90)
print(resultado3)
4. Middleware HTTP pra APIs
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
store = IdempotencyStore("api_idempotency.db")
class IdempotentAPIHandler(BaseHTTPRequestHandler):
def do_POST(self):
idem_key = self.headers.get("Idempotency-Key")
if not idem_key:
self.send_error(400, "Header 'Idempotency-Key' é obrigatório")
return
cached = store.get(idem_key)
if cached:
result, status = cached
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("X-Idempotent-Replayed", "true")
self.end_headers()
self.wfile.write(json.dumps(result).encode())
return
if not store.reserve(idem_key):
self.send_error(409, "Requisição duplicada em processamento")
return
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length)
try:
result = self._process_request(body)
store.complete(idem_key, result, 201)
self.send_response(201)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(result).encode())
except Exception as e:
store.complete(idem_key, {"error": str(e)}, 500)
self.send_error(500, str(e))
def _process_request(self, body: bytes) -> dict:
data = json.loads(body)
return {
"id": "rec_" + idem_key[:8],
"data": data,
"created_at": datetime.now().isoformat()
}
Testando o Sistema
store = IdempotencyStore("test.db")
@idempotent(store)
def operacao_lenta(id: str):
import time
time.sleep(1)
return {"id": id, "status": "ok"}
r1 = operacao_lenta("op_001")
print(f"Primeira: {r1}")
r2 = operacao_lenta("op_001")
print(f"Segunda (cache): {r2}")
store.cleanup_expired()
print("Keys expiradas removidas")
Boas Práticas (Que Eu Aprendi na Marra)
1. Sempre use TTL. Keys eternas lotam o banco. 24 horas é um bom padrão, mas ajuste conforme seu caso de uso.
2. Hash a key antes de armazenar. Se a key contiver dados sensíveis (como CPF ou e-mail), você não quer isso em texto puro no banco.
3. Reserve antes de executar. Isso impede race conditions quando duas requisições com a mesma key chegam quase simultaneamente.
4. Retorne o status code original. Se a primeira requisição retornou 201, o replay também deve retornar 201, não 200.
5. Adicione um header de replay. X-Idempotent-Replayed: true ajuda no debug e logs.
6. Trate erros corretamente. Se a operação falhar, armazene o erro também. Assim, retries subsequentes retornam o mesmo erro em vez de executar de novo.
Quando NÃO Usar Idempotency Keys
Nem toda operação precisa de idempotência. Operações de leitura (GET) são naturalmente idempotentes. Operações que geram valores únicos (como IDs sequenciais) podem ficar estranhas com replay.
Use idempotência pra:
- Pagamentos e cobranças
- Criação de recursos críticos
- Envio de e-mails e notificações
- Qualquer operação com efeitos colaterais externos
Qual Operação da Sua Automação Merece Idempotência?
Agora é sua vez: qual parte do seu código você tem medo de executar duas vezes? Comenta aí embaixo que eu te ajudo a pensar na estratégia de key certa pro seu caso.
