FastAPI: endpoints idempotentes com Idempotency-Key e Redis (retries seguros)

Contexto e motivacao

Falhas de rede, timeouts e retries automáticos (cliente ou infra) fazem com que operações não-idempotentes sejam executadas várias vezes. Isso gera cobranças duplicadas, criação de recursos em duplicidade e inconsistência de estado. Usar um cabeçalho Idempotency-Key junto com um armazenamento rápido como Redis permite detectar retries do mesmo pedido e garantir que a ação seja executada apenas uma vez ou que devolvamos uma resposta consistente.

Desenvolvedores no nível intermediário normalmente conhecem FastAPI e Redis, mas sofrem com a implementação correta do fluxo de idempotência em ambiente concorrente e com múltiplos workers. O objetivo aqui é entregar um padrão robusto que funciona em produção: bloqueio idempotente, armazenamento de resposta, expiração e proteção contra race conditions.

O padrão se aplica a endpoints que criam recursos ou disparam operações latentes (pagamentos, filas, geração de relatórios). Não é necessário para GETs, mas é crítico em POSTs/PUTs que não podem ser duplicados.

Este artigo explica o fluxo, detalha a implementação usando FastAPI + redis.asyncio (redis-py 4.x), mostra scripts Lua para atomicidade e oferece padrões operacionais e armadilhas a evitar.

Como funciona por baixo dos panos

Idempotência via chave consiste em vários passos: identificar a solicitação com a chave fornecida, verificar se já existe resultado, reservar um marcador de processamento (lock) se for a primeira vez, executar a operação e persistir o resultado para chamadas subsequentes. Tudo isso precisa ser coordenado de forma atômica para evitar duas execuções concorrentes para a mesma chave.

Estados e transições

Usamos três estados principais armazenados no Redis:

• in-progress: existe um lock indicando que um worker está processando a requisição.
• completed: a operação terminou com resultado persistido.
• absent: não existe registro para a chave.

Transições importantes:

1. absent -> in-progress: realizado via SET NX com TTL (claim).
2. in-progress -> completed: substituição atômica que grava resposta e remove o lock.
3. in-progress -> absent: se o worker morreu e o TTL expirar, outra tentativa pode re-entrar.

Atomicidade e segurança

Operações simultâneas são o maior risco. Usamos SET NX (ou script Lua) para garantir que apenas um worker faça o trabalho inicial. Para gravar a resposta final, verificamos se a operação ainda pertence ao mesmo owner (uuid) antes de substituir o estado. Isso evita que um worker sobrescreva o resultado de outro caso o lock tenha expirado e outra tentativa tenha assumido.

Use sempre um identificador de owner por lock (UUID) e valide-o ao persistir o resultado. Scripts Lua ajudam a garantir essas checagens atomicas.

Implementacao passo a passo

A implementação que vamos usar combina um helper de Redis, um decorator para endpoints e um padrão de serialização de resposta. O código depende de redis.asyncio e FastAPI; para serializar use orjson para desempenho.

Dependências e configurações

Instale as dependências recomendadas:

pip install fastapi uvicorn redis[asyncio] orjson python-dotenv

Configuração mínima (exemplo .env):

REDIS_URL=redis://localhost:6379/0
IDEMPOTENCY_TTL=3600  # segundos

O TTL evita que chaves ocupem memória indefinidamente e serve como proteção contra keys abandonadas.

Helper Redis e script Lua (atomicidade)

O próximo bloco cria helpers para claim, salvar resposta e recuperar. Ele usa um script Lua para setar o estado final somente se o owner bater — isso é atômico no Redis.

"""redis_helpers.py
Helpers async para gerenciar locks idempotentes e armazenamento de resposta.
"""
import asyncio
import uuid
import json
from typing import Optional
import redis.asyncio as aioredis

IDEMPOTENCY_PREFIX = "idem:"  # chave no redis

# Script Lua: grava resposta somente se owner bate
LUA_SET_COMPLETED = """
if redis.call('GET', KEYS[1]) == ARGV[1] then
  redis.call('SET', KEYS[2], ARGV[2], 'PX', ARGV[3])
  redis.call('DEL', KEYS[1])
  return 1
else
  return 0
end
"""

class RedisIdempotency:
    def __init__(self, url: str, ttl: int = 3600):
        self.redis = aioredis.from_url(url, decode_responses=True)
        self.ttl = ttl * 1000  # ms for SET px
        self._script = None

    async def _load_script(self):
        if self._script is None:
            self._script = await self.redis.script_load(LUA_SET_COMPLETED)
        return self._script

    async def claim(self, key: str, owner: Optional[str] = None, claim_ttl: int = 30000) -> str:
        """Tenta criar um lock in-progress. Retorna owner usado (novo ou existente).
        claim_ttl em ms.
        """
        if owner is None:
            owner = str(uuid.uuid4())
        lock_key = IDEMPOTENCY_PREFIX + key + ":lock"
        # SET lock value NX PX claim_ttl
        ok = await self.redis.set(lock_key, owner, nx=True, px=claim_ttl)
        if ok:
            return owner
        # se não conseguiu, retorna o owner existente
        existing = await self.redis.get(lock_key)
        return existing

    async def set_completed(self, key: str, owner: str, response_obj: dict, store_ttl: Optional[int] = None) -> bool:
        """Tenta gravar o resultado final se owner bater. Retorna True se bem sucedido."""
        if store_ttl is None:
            store_ttl = self.ttl
        completed_key = IDEMPOTENCY_PREFIX + key + ":resp"
        lock_key = IDEMPOTENCY_PREFIX + key + ":lock"
        response_json = json.dumps(response_obj, separators=(",",":"))
        script = await self._load_script()
        res = await self.redis.evalsha(script, 2, lock_key, completed_key, owner, response_json, str(store_ttl))
        return res == 1

    async def get_completed(self, key: str) -> Optional[dict]:
        completed_key = IDEMPOTENCY_PREFIX + key + ":resp"
        raw = await self.redis.get(completed_key)
        if not raw:
            return None
        return json.loads(raw)

    async def wait_for_completion(self, key: str, timeout: float = 5.0, poll_interval: float = 0.1) -> Optional[dict]:
        """Se existir in-progress, espera até que completed apareça ou timeout."""
        elapsed = 0.0
        while elapsed < timeout:
            res = await self.get_completed(key)
            if res is not None:
                return res
            await asyncio.sleep(poll_interval)
            elapsed += poll_interval
        return None

O que aconteceu: o helper expõe claim (SET NX), set_completed rodando um script Lua que grava resposta apenas se o lock pertencer ao owner, e um get/espera para recuperar a resposta existente. Isso reduz races.

Decorator idempotente para endpoints

Agora um decorator que adiciona o comportamento idempotente ao handler. Ele:

• lê cabeçalho Idempotency-Key
• verifica se já existe resposta completa e a retorna
• tenta claim do lock; se não conseguir espera por resultado
• executa handler, captura resposta (status, headers, body) e persiste

"""decorators.py
Decorator idempotente para FastAPI endpoints.
"""
from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
import orjson
from functools import wraps
from typing import Callable

from redis_helpers import RedisIdempotency

redis_client = RedisIdempotency(url="redis://localhost:6379/0", ttl=3600)

def idempotent_endpoint(func: Callable):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        # procura Request em args/kwargs
        request: Request = kwargs.get('request') or next((a for a in args if isinstance(a, Request)), None)
        if request is None:
            raise RuntimeError('idempotent_endpoint requires a Request parameter')

        key = request.headers.get('Idempotency-Key')
        if not key:
            raise HTTPException(status_code=400, detail='Idempotency-Key header required')

        # 1) se já existe completed -> retorna
        existing = await redis_client.get_completed(key)
        if existing is not None:
            return JSONResponse(content=existing['body'], status_code=existing['status'], headers=existing.get('headers'))

        # 2) tenta claim
        owner = await redis_client.claim(key)
        # se não somos owner, espera que outro complete
        if owner is not None:
            # se owner não é nosso uuid, significa que outro está processando
            # espera curto
            waited = await redis_client.wait_for_completion(key, timeout=5.0)
            if waited is not None:
                return JSONResponse(content=waited['body'], status_code=waited['status'], headers=waited.get('headers'))
            # se timeout -> conflito
            raise HTTPException(status_code=409, detail='Processing in progress; try again later')

        # 3) somos owner: executa handler
        result = await func(*args, **kwargs)
        # normaliza resposta (assume JSON serializavel)
        if isinstance(result, JSONResponse):
            body = result.body.decode() if isinstance(result.body, (bytes, bytearray)) else result.body
            status = result.status_code
            headers = dict(result.headers)
        elif isinstance(result, dict):
            body = result
            status = 200
            headers = {}
        else:
            # tenta serializar genericamente
            body = result
            status = 200
            headers = {}

        # 4) grava completed (owner validation feita no script)
        payload = {'status': status, 'headers': headers, 'body': body}
        ok = await redis_client.set_completed(key, owner, payload)
        if not ok:
            # alguém tomou o lock antes do set (edge case)
            # tenta recuperar o resultado do novo owner
            final = await redis_client.get_completed(key)
            if final:
                return JSONResponse(content=final['body'], status_code=final['status'], headers=final.get('headers'))
            raise HTTPException(status_code=500, detail='Idempotency conflict')

        return JSONResponse(content=body, status_code=status, headers=headers)

    return wrapper

Depois desse bloco, endpoints decorados com @idempotent_endpoint terão comportamento consistente. O decorator assume que handler retorna dict ou JSONResponse. Você pode adaptar para Response completa ou streams se necessário.

Exemplo de endpoint real

Exemplo de um endpoint de cobrança que usa o decorator. Simula operação que demora e que não deve rodar duas vezes.

from fastapi import FastAPI, Request
from decorators import idempotent_endpoint
import asyncio

app = FastAPI()

@app.post('/charge')
@idempotent_endpoint
async def create_charge(request: Request):
    # simula chamada a gateway de pagamentos que demora
    await asyncio.sleep(1.5)
    # normalmente aqui você chamaria um serviço externo e gravaria no banco
    result = {'transaction_id': 'tx-' + request.headers.get('Idempotency-Key'), 'status': 'success'}
    return result

O que acontece: na primeira chamada com a chave X o decorator grava lock e executa create_charge. Chamadas repetidas com a mesma chave X retornam imediatamente a resposta persistida sem chamar o handler novamente.

Patterns e boas praticas

Algumas decisões de implementação impactam diretamente a robustez do sistema. Abaixo, patterns empíricos usados em produção.

Chave: escolha e validade

Recomende aos consumidores que gerem Idempotency-Key como UUID v4. Evite usar dados sensíveis ou previsíveis. Documente o TTL máximo suportado (ex: 24h) para que clientes saibam por quanto tempo podem reenviar a mesma chave com segurança.

TTL e expiracao

Configurar TTL para lock e para objeto completado é crítico:

• Lock TTL: deve cobrir o tempo máximo esperado da operação + margem.
• Resultado TTL: tempo que você quer permitir retries idempotentes. Cobrir alguns dias para operações financeiras pode ser necessário.

Regra prática: lock_ttl << result_ttl. Lock curto evita bloqueios permanentes por worker morto; resultado longo permite retries sem reprocessamento.

Garantir não repetição no banco externo

Mesmo com resposta persistida, a sua operação pode ter side-effects externos (banco, gateway). Combine persistência do resultado com um único ponto de escrita no banco:

• Trate a gravação do recurso como idempotente no DB (unique constraints, idempotency_key column).
• Se gravar no DB, escreva o resultado junto da idempotency_key numa mesma transação para garantir consistência.

Armadilhas comuns e como evitar

Aqui estão erros frequentes e como mitigá-los com implementações práticas.

1) Não validar owner antes de gravar

Problema: um worker perde o lock por timeout e outro assume; o primeiro grava resultado depois e sobrescreve o correto. Solução: sempre gravar final somente se owner bater (script Lua acima) ou usando WATCH/MULTI com checagens atômicas.

2) Tratar idempotency-key como segredo

Problema: chaves expostas podem permitir replay por terceiros. Solução: trate como token não-sensível, monitore anomalias e aplique rate-limits por client-id/IP. Para operações sensíveis, combine Idempotency-Key com autenticação forte.

3) Bloqueio indefinido por lock muito longo

Problema: se um worker trava e lock TTL é enorme, as requisições subsequentes ficarão bloqueadas ou falharão. Solução: use TTLs moderados, implemente wait_with_timeout e retorne 409/202 apropriados para quem tenta enquanto ainda está em processamento.

4) Serializacao de resposta inconsistente

Problema: serializar objetos complexos (bytes, datetimes) pode causar diferenças entre primeira execução e recuperações. Solução: normalize a resposta ao gravar — converta datetimes para ISO, remova campos transitórios e defina contrato claro de resposta armazenada.

Dica: logue o Idempotency-Key + owner + request-id para rastrear execuções e diagnosticar condicionais de corrida.

Armadilhas operacionais e testes

Teste com concorrência: scripts que disparam 50-200 requests simultâneos com a mesma chave são essenciais para validar o comportamento. Simule falhas de worker desligando processos durante execução.

Monitore métricas chave: número de keys expiradas, hits no completed, número de conflitos (409), tempo médio de espera para completed. Configure alertas para spikes incomuns.

Conclusao pratica

Idempotency-Key + Redis é uma solução robusta se implementada corretamente: claim com SET NX, validação de owner e gravação atômica do resultado. Use TTLs bem dimensionados, unique constraints no banco para compatibilidade dupla e scripts Lua quando precisar garantir atomicidade entre checagens e gravações.

Implemente o decorator/ middleware mostrado como base e adapte para o contrato de resposta real da sua aplicação. Teste com concorrência e falhas de infraestrutura. Documente o comportamento esperado para consumidores do API — TTL, formato da chave e semântica de retries.

Pontos finais: sempre trate idempotency como uma característica de API, não apenas como um hack de infra; combine armazenamento de resultado com garantias no backend e logue para auditoria.

Com esse padrão você reduz significativamente riscos de duplicação, melhora a experiência do cliente em redes instáveis e habilita retries seguros no frontend e infra.