Como implementar caching distribuído para consultas de CPF com Redis

Implementar caching com Redis para consultas de CPF reduz chamadas à API em até 90%, entregando resultados em menos de 1ms para CPFs já consultados — contra os ~900ms de uma chamada direta à API. O padrão consiste em verificar o Redis antes de cada consulta, armazenar o resultado com um TTL adequado ao caso de uso e nunca cachear respostas de erro. Quando sua aplicação consulta os mesmos CPFs repetidamente, o Redis é a camada mais eficiente para economizar cota e reduzir latência.

Por que usar caching para consultas de CPF

Existem três razões principais para implementar cache em consultas de CPF.

Economia de consultas

Cada plano da CPFHub.io inclui um número de consultas por mês. O plano gratuito oferece 50, o Pro oferece 1.000. Ao cachear resultados, você evita consultas duplicadas e aproveita melhor o limite disponível.

Performance

Mesmo com o tempo de resposta de ~900ms da CPFHub.io, uma consulta cacheada no Redis retorna em menos de 1ms. Em fluxos que exigem resposta instantânea, como checkout de e-commerce, essa diferença é significativa.

Resiliência

Se a API estiver temporariamente indisponível, os dados cacheados permitem que sua aplicação continue funcionando para CPFs já consultados.

Arquitetura do caching

O fluxo básico de caching para consultas de CPF segue este padrão:

1. A aplicação recebe uma solicitação de consulta de CPF.
2. Verifica se o resultado existe no Redis.
3. Se existir (cache hit), retorna o dado cacheado.
4. Se não existir (cache miss), consulta a API da CPFHub.io.
5. Armazena o resultado no Redis com um TTL (time-to-live).
6. Retorna o resultado ao solicitante.

Implementação em Python

import redis
import requests
import json
import hashlib

# Configuração do Redis
redis_client = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

CPFHUB_API_KEY = 'SUA_CHAVE_DE_API'
CACHE_TTL_SECONDS = 3600 # 1 hora

def gerar_chave_cache(cpf: str) -> str:
    cpf_limpo = cpf.replace('.', '').replace('-', '')
    return f'cpfhub:consulta:{cpf_limpo}'

def consultar_cpf_com_cache(cpf: str) -> dict:
    chave = gerar_chave_cache(cpf)

    # 1. Verificar cache
    resultado_cache = redis_client.get(chave)
    if resultado_cache:
    dados = json.loads(resultado_cache)
    dados['_cache'] = True
    return dados

    # 2. Consultar API
    cpf_limpo = cpf.replace('.', '').replace('-', '')
    url = f'https://api.cpfhub.io/cpf/{cpf_limpo}'
    headers = {
    'x-api-key': CPFHUB_API_KEY,
    'Accept': 'application/json'
    }

    response = requests.get(url, headers=headers, timeout=10)
    resultado = response.json()

    # 3. Armazenar no cache apenas respostas com sucesso
    if resultado.get('success'):
    redis_client.setex(
    chave,
    CACHE_TTL_SECONDS,
    json.dumps(resultado)
    )

    resultado['_cache'] = False
    return resultado

# Exemplo de uso
resultado1 = consultar_cpf_com_cache('12345678900')
print(f'Cache: {resultado1["_cache"]}') # False (primeira consulta)

resultado2 = consultar_cpf_com_cache('12345678900')
print(f'Cache: {resultado2["_cache"]}') # True (cache hit)

Implementação em Node.js

const Redis = require('ioredis');

const redis = new Redis({
    host: 'localhost',
    port: 6379
});

const CPFHUB_API_KEY = 'SUA_CHAVE_DE_API';
const CACHE_TTL_SECONDS = 3600;

async function consultarCpfComCache(cpf) {
    const cpfLimpo = cpf.replace(/\D/g, '');
    const chaveCache = `cpfhub:consulta:${cpfLimpo}`;

    // 1. Verificar cache
    const cacheado = await redis.get(chaveCache);
    if (cacheado) {
    const dados = JSON.parse(cacheado);
    dados._cache = true;
    return dados;
    }

    // 2. Consultar API
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), 10000);

    try {
    const response = await fetch(
    `https://api.cpfhub.io/cpf/${cpfLimpo}`,
    {
    headers: {
    'x-api-key': CPFHUB_API_KEY,
    'Accept': 'application/json'
    },
    signal: controller.signal
    }
    );

    clearTimeout(timeoutId);
    const resultado = await response.json();

    // 3. Armazenar no cache
    if (resultado.success) {
    await redis.setex(chaveCache, CACHE_TTL_SECONDS, JSON.stringify(resultado));
    }

    resultado._cache = false;
    return resultado;
    } catch (error) {
    clearTimeout(timeoutId);
    throw error;
    }
}

// Exemplo de uso
(async () => {
    const resultado1 = await consultarCpfComCache('12345678900');
    console.log(`Cache: ${resultado1._cache}`); // false

    const resultado2 = await consultarCpfComCache('12345678900');
    console.log(`Cache: ${resultado2._cache}`); // true

    await redis.quit();
})();

Definindo o TTL ideal

O TTL (time-to-live) determina por quanto tempo o resultado fica no cache antes de expirar. A escolha do TTL depende do caso de uso.

Invalidação de cache

Em alguns cenários, é necessário invalidar o cache antes do TTL expirar.

Invalidação por CPF específico

def invalidar_cache_cpf(cpf: str):
    chave = gerar_chave_cache(cpf)
    redis_client.delete(chave)
    print(f'Cache invalidado para CPF: {cpf}')

Invalidação em massa

def invalidar_todo_cache():
    chaves = redis_client.keys('cpfhub:consulta:*')
    if chaves:
    redis_client.delete(*chaves)
    print(f'{len(chaves)} entradas de cache removidas')

Métricas de cache

Para monitorar a eficácia do cache, registre as taxas de hit e miss.

import time

def consultar_cpf_com_metricas(cpf: str) -> dict:
    chave = gerar_chave_cache(cpf)
    inicio = time.time()

    resultado_cache = redis_client.get(chave)

    if resultado_cache:
    tempo = (time.time() - inicio) * 1000
    redis_client.incr('cpfhub:metricas:cache_hits')
    dados = json.loads(resultado_cache)
    dados['_cache'] = True
    dados['_tempo_ms'] = round(tempo, 2)
    return dados

    # Cache miss -- consultar API
    redis_client.incr('cpfhub:metricas:cache_misses')

    cpf_limpo = cpf.replace('.', '').replace('-', '')
    url = f'https://api.cpfhub.io/cpf/{cpf_limpo}'
    headers = {
    'x-api-key': CPFHUB_API_KEY,
    'Accept': 'application/json'
    }

    response = requests.get(url, headers=headers, timeout=10)
    tempo = (time.time() - inicio) * 1000
    resultado = response.json()

    if resultado.get('success'):
    redis_client.setex(chave, CACHE_TTL_SECONDS, json.dumps(resultado))

    resultado['_cache'] = False
    resultado['_tempo_ms'] = round(tempo, 2)
    return resultado

def obter_metricas() -> dict:
    hits = int(redis_client.get('cpfhub:metricas:cache_hits') or 0)
    misses = int(redis_client.get('cpfhub:metricas:cache_misses') or 0)
    total = hits + misses

    return {
    'hits': hits,
    'misses': misses,
    'total': total,
    'taxa_hit': f'{hits/total:.1%}' if total > 0 else '0%'
    }

Boas práticas

• Nunca cachear erros — Apenas armazene no cache respostas com success: true. Erros temporários da API não devem ser servidos do cache.

• Use namespaces nas chaves — Prefixe as chaves com cpfhub:consulta: para facilitar a identificação e a invalidação seletiva.

• Configure limites de memória no Redis — Defina uma política de eviction (como allkeys-lru) para evitar que o Redis consuma toda a memória do servidor. A documentação oficial do Redis detalha todas as políticas disponíveis.

• Considere a LGPD — Dados de CPF armazenados em cache são dados pessoais. Garanta que o Redis esteja protegido com autenticação e criptografia.

• Monitore a taxa de hit — Uma taxa de hit abaixo de 50% indica que o cache não está sendo efetivo. Revise o TTL ou a estratégia de caching.

Perguntas frequentes

Qual TTL devo usar para cache de CPF em um sistema de onboarding?

Para onboarding, o TTL ideal fica entre 30 minutos e 1 hora. Dados cadastrais de CPF raramente mudam em janelas curtas, mas um TTL longo demais pode gerar inconsistências se o status do CPF for alterado na Receita Federal. Para KYC em fintechs, onde a atualidade dos dados é crítica, prefira 30 minutos e force a revalidação em operações de alto valor.

O Redis pode ser usado em produção para dados de CPF sem violar a LGPD?

Sim, desde que o Redis esteja configurado corretamente. Habilite autenticação com senha forte (requirepass), TLS para conexões em trânsito e, se possível, criptografia em repouso. Defina uma política de eviction (allkeys-lru) e restrinja o acesso por IP. O cache de CPF é legítimo sob a LGPD quando existe base legal para o tratamento e o dado é protegido adequadamente.

O cache de CPF reduz custos reais na prática?

Depende do perfil de consultas da aplicação. Em e-commerce com cadastro único por CPF, o ganho é menor. Em sistemas onde o mesmo CPF é validado em múltiplas etapas do funil (cadastro, checkout, recarga), o cache pode eliminar 70–90% das chamadas à API. Com o plano Pro da CPFHub.io (1.000 consultas/mês, R$149), cada consulta cacheada economiza R$0,15 em excedente.

Como evitar que dados desatualizados sejam servidos pelo cache?

Use TTLs curtos para dados críticos e implemente invalidação manual nos eventos que possam alterar o estado do CPF (como uma notificação de irregularidade). Para compliance, considere sempre consultar a API diretamente em operações que envolvam valores altos ou que exijam KYC atualizado, usando o cache apenas para etapas preliminares do fluxo.

Conclusão

Implementar caching distribuído com Redis para consultas de CPF é uma das formas mais eficazes de otimizar a performance e reduzir custos na integração com APIs externas. Com a estratégia certa de TTL e invalidação, é possível servir a maioria das consultas em menos de 1ms, reservando as chamadas reais à API da CPFHub.io para os momentos em que os dados precisam ser frescos.

A combinação de cache bem configurado com boas práticas de LGPD — autenticação, TLS e TTLs adequados — garante que a solução seja ao mesmo tempo performática, econômica e em conformidade regulatória. O resultado prático é menos latência para o usuário final e menor custo por consulta ao longo do tempo.

Cadastre-se em cpfhub.io e comece gratuitamente — 50 consultas por mês, sem cartão de crédito.