API de CPF: como garantir idempotencia em chamadas repetidas

Para garantir idempotência em chamadas repetidas à API de CPF, implemente cache com TTL no lado do cliente — armazenando o resultado da primeira consulta e reutilizando-o em chamadas subsequentes com o mesmo identificador. Assim, falhas de rede e retentativas automáticas não geram consultas duplicadas nem desperdiçam cota. A OWASP recomenda essa abordagem como boa prática em integrações resilientes.

Introdução

Em sistemas distribuidos, falhas de rede, timeouts e retentativas automáticas podem causar chamadas duplicadas a uma API. Quando isso acontece com consultas de CPF, o risco é consumir cota desnecessariamente e processar o mesmo resultado mais de uma vez. A solução para esse problema é a idempotência.

O que é idempotência

Uma operação é idempotente quando executá-la uma ou múltiplas vezes produz exatamente o mesmo resultado. No contexto de APIs:

• GET — Naturalmente idempotente. Consultar o mesmo CPF duas vezes retorna os mesmos dados.

• POST — Não é naturalmente idempotente. Sem cuidado, pode criar registros duplicados.

A API da CPFHub.io utiliza o método GET, que é inerentemente idempotente do lado do servidor. No entanto, do lado do cliente, cada chamada consome uma consulta da sua cota mensal. Por isso, a idempotência no cliente é essencial para evitar desperdício.

Por que idempotência importa em consultas de CPF

• Economia de cota — Cada consulta duplicada gasta uma unidade do seu plano (50/mês no Gratuito, 1.000/mês no Pro).

• Consistência de dados — Evita processar o mesmo CPF duas vezes em fluxos de onboarding.

• Resiliência — Permite retentativas seguras em caso de falha de rede.

• Auditoria — Facilita rastrear cada consulta de forma única.

Estratégia 1: Idempotency key

Gere um identificador único para cada operação lógica e armazene o resultado:

import requests
import hashlib
import json
import redis
import uuid

r = redis.Redis(host='localhost', port=6379, db=0)

def consultar_cpf_idempotente(cpf, operacao_id=None):
    # Gerar chave de idempotência baseada no CPF e na operação
    if operacao_id is None:
    operacao_id = str(uuid.uuid4())

    idempotency_key = f'cpf_consulta:{cpf}:{operacao_id}'

    # Verificar se já existe resultado em cache
    resultado_cache = r.get(idempotency_key)
    if resultado_cache:
    print(f'Cache hit para {idempotency_key}')
    return json.loads(resultado_cache)

    # Fazer a consulta real
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
    }
    response = requests.get(url, headers=headers, timeout=15)
    resultado = response.json()

    # Armazenar resultado com TTL de 1 hora
    r.setex(idempotency_key, 3600, json.dumps(resultado))

    return resultado

Com essa abordagem, mesmo que a função seja chamada múltiplas vezes com o mesmo operacao_id, a API real é chamada apenas uma vez.

Estratégia 2: Cache por CPF com TTL

Para cenários onde o mesmo CPF pode ser consultado por diferentes fluxos em curto intervalo:

const Redis = require('ioredis');
const redis = new Redis();

async function consultarCPF(cpf) {
    const cacheKey = `cpf:${cpf}`;

    // Verificar cache
    const cached = await redis.get(cacheKey);
    if (cached) {
    return JSON.parse(cached);
    }

    // Consulta real
    const response = await fetch(`https://api.cpfhub.io/cpf/${cpf}`, {
    method: 'GET',
    headers: {
    'x-api-key': process.env.CPFHUB_API_KEY,
    'Accept': 'application/json'
    },
    signal: AbortSignal.timeout(15000)
    });

    const data = await response.json();

    // Cache por 30 minutos
    await redis.setex(cacheKey, 1800, JSON.stringify(data));

    return data;
}

Estratégia 3: Deduplicação com lock distribuído

Quando múltiplas instâncias do serviço podem tentar consultar o mesmo CPF simultaneamente, use um lock:

import requests
import json
import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def consultar_cpf_com_lock(cpf):
    cache_key = f'cpf:resultado:{cpf}'
    lock_key = f'cpf:lock:{cpf}'

    # Verificar cache primeiro
    resultado_cache = r.get(cache_key)
    if resultado_cache:
    return json.loads(resultado_cache)

    # Tentar adquirir lock (TTL de 30 segundos)
    lock_adquirido = r.set(lock_key, '1', nx=True, ex=30)

    if not lock_adquirido:
    # Outra instância está consultando, esperar pelo resultado
    for _ in range(30):
    time.sleep(1)
    resultado_cache = r.get(cache_key)
    if resultado_cache:
    return json.loads(resultado_cache)
    raise TimeoutError('Timeout aguardando resultado de consulta em andamento')

    try:
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
    }
    response = requests.get(url, headers=headers, timeout=15)
    resultado = response.json()

    # Armazenar no cache por 1 hora
    r.setex(cache_key, 3600, json.dumps(resultado))

    return resultado
    finally:
    r.delete(lock_key)

Estratégia 4: Registro em banco de dados

Para auditoria completa, registre cada consulta única no banco:

import requests
import sqlite3
from datetime import datetime

def consultar_cpf_com_registro(cpf, motivo, solicitante):
    conn = sqlite3.connect('consultas_cpf.db')
    cursor = conn.cursor()

    # Verificar se já existe consulta recente (últimas 24h)
    cursor.execute('''
    SELECT resultado FROM consultas
    WHERE cpf = ? AND timestamp > datetime('now', '-24 hours')
    ORDER BY timestamp DESC LIMIT 1
    ''', (cpf,))

    row = cursor.fetchone()
    if row:
    conn.close()
    return {'fonte': 'cache_db', 'data': row[0]}

    # Fazer consulta real
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': 'SUA_CHAVE_DE_API',
    'Accept': 'application/json'
    }
    response = requests.get(url, headers=headers, timeout=15)
    resultado = response.json()

    # Registrar consulta
    cursor.execute('''
    INSERT INTO consultas (cpf, motivo, solicitante, resultado, timestamp)
    VALUES (?, ?, ?, ?, ?)
    ''', (cpf, motivo, solicitante, str(resultado), datetime.utcnow().isoformat()))

    conn.commit()
    conn.close()
    return {'fonte': 'api', 'data': resultado}

Comparação das estratégias

Definindo o TTL ideal

O tempo de vida do cache depende do seu caso de uso:

• Onboarding (cadastro) — TTL de 5 a 15 minutos. O usuário pode tentar novamente.

• Verificação periódica — TTL de 24 horas. Dados cadastrais não mudam com frequência.

• Checkout em tempo real — TTL de 1 a 5 minutos. Priorizar dados frescos.

Perguntas frequentes

Como a idempotência reduz o consumo de cota na API de CPF?

Ao armazenar o resultado da primeira consulta em cache (Redis, banco de dados ou memória), chamadas subsequentes com o mesmo identificador retornam o dado local sem acionar a API. Isso é especialmente útil em fluxos com retentativas automáticas: em vez de gastar múltiplas unidades de cota, a integração consome apenas uma por operação lógica.

O que acontece se o limite de consultas do plano for atingido?

A API da CPFHub.io não bloqueia chamadas ao atingir o limite do plano. Cada consulta excedente é cobrada a R$0,15, sem interrupção do serviço. Implementar idempotência reduz o risco de acúmulo de excedente em cenários de retentativas.

Qual estratégia de idempotência é mais simples para começar?

O cache por CPF com TTL em Redis é o ponto de entrada mais direto: poucas linhas de código, sem necessidade de gerenciar locks ou banco relacional. Para a maioria dos fluxos de onboarding, um TTL de 5 a 30 minutos já elimina a maioria das consultas duplicadas.

Como garantir conformidade com a LGPD ao cachear dados de CPF?

Armazene apenas o necessário — nome e status, não o CPF em texto plano se um token bastar. Defina TTL curto (máximo 24h para dados de verificação), implemente controle de acesso ao Redis e documente a base legal para o tratamento. A ANPD orienta que dados de identificação devem ser tratados com o princípio da necessidade.

Conclusão

Implementar idempotência em consultas de CPF protege sua cota, evita processamento duplicado e torna sua integração mais resiliente. A combinação de cache com TTL, locks distribuídos e registros de auditoria cria uma camada robusta que funciona em qualquer escala.

Cadastre-se em cpfhub.io — 50 consultas mensais gratuitas, sem cartão de crédito.