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

> Aprenda a implementar caching distribuído com Redis para consultas de CPF via API, reduzindo custos e melhorando a performance da sua aplicação.

**Publicado:** 14/04/2026
**Autor:** Redação CPFHub.io
**URL:** https://cpfhub.io/blog/como-implementar-caching-distribuido-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

```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

```javascript
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.

| Caso de uso | TTL recomendado | Justificativa |
| --- | --- | --- |
| Onboarding de clientes | 1 hora | Dados precisam ser recentes |
| Checkout de e-commerce | 30 minutos | Velocidade é prioridade |
| Validação em batch | 24 horas | CPFs consultados em lote raramente mudam no mesmo dia |
| Consulta recorrente do mesmo cliente | 6 horas | Equilíbrio entre atualidade e economia |
| Auditoria e compliance | Sem cache | Dados devem ser sempre atualizados |

---

## Invalidação de cache

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

### Invalidação por CPF específico

```python
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

```python
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.

```python
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](https://redis.io/docs/latest/develop/reference/eviction/) 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.

### Leia também

- [SLA de API de CPF: entenda os níveis de disponibilidade](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade)
- [Como garantir alta disponibilidade na API de CPF](https://cpfhub.io/blog/como-garantir-alta-disponibilidade-na-api-de-cpf)
- [Como implementar cache inteligente para respostas da API de CPF](https://cpfhub.io/blog/como-implementar-cache-inteligente-respostas-api-cpf)
- [Como monitorar consumo e custos da API de CPF com observabilidade](https://cpfhub.io/blog/api-cpf-monitorar-consumo-custos-observabilidade)

---

## 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**](https://www.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](https://www.cpfhub.io/) e comece gratuitamente — 50 consultas por mês, sem cartão de crédito.