# Como Evitar Quedas de Serviço ao Usar uma API Gratuita de CPF > Aprenda a evitar quedas de serviço ao usar uma API gratuita de CPF. Estratégias de cache, fallback, circuit breaker e degradação graciosa. **Publicado:** 01/02/2025 **Autor:** Redação CPFHub.io **URL:** https://cpfhub.io/blog/evitar-quedas-servico-api-gratuita-cpf --- Para evitar quedas de serviço ao usar uma API gratuita de CPF, implemente quatro camadas em sequência: cache local com TTL de 24 horas para CPFs já consultados, timeout de 10 segundos por requisição, circuit breaker que ativa fallback após 5 falhas consecutivas e validação algorítmica local como último recurso. Essas estratégias mantêm o sistema funcional mesmo quando a API está temporariamente indisponível. ## Introdução APIs gratuitas oferecem um excelente custo-benefício, mas tipicamente não vêm com garantias formais de disponibilidade (SLA). Isso não significa que seu sistema precisa ficar vulnerável a indisponibilidades. Com as estratégias certas de resiliência, cache, fallback e degradação graciosa, você pode construir um sistema que funciona de forma confiável mesmo quando a API está temporariamente fora do ar. --- ## Camadas de proteção contra indisponibilidade A resiliência é construída em camadas, onde cada uma protege contra um tipo diferente de falha. | Camada | Proteção contra | Implementação | |--------|----------------|---------------| | **Cache local** | Requisições repetidas | In-memory ou Redis | | **Timeout** | API lenta ou travada | Timeout de 5-10s por requisição | | **Retry com backoff** | Falhas transitórias | 3 tentativas com delay exponencial | | **Circuit breaker** | API completamente fora | Estado aberto após N falhas | | **Fallback local** | Todas as falhas de rede | Validação algorítmica do CPF | | **Fila assíncrona** | Picos de demanda | Processar em background | - **Defesa em profundidade** -- cada camada captura falhas que passaram pela anterior - **Graceful degradation** -- o sistema oferece funcionalidade reduzida em vez de falhar completamente - **Transparência** -- o usuário recebe feedback sobre o nível de validação aplicado --- ## Implementando cache para reduzir dependência Cache é a primeira e mais eficaz camada de proteção. CPFs consultados recentemente não precisam ir à API. ```python import time import hashlib from collections import OrderedDict import requests import os class CPFCache: def __init__(self, max_entries: int = 10000, ttl_seconds: int = 86400): self.cache = OrderedDict() self.max_entries = max_entries self.ttl = ttl_seconds self.hits = 0 self.misses = 0 def get(self, cpf: str): chave = self._hash(cpf) if chave in self.cache: entrada = self.cache[chave] if time.time() - entrada["timestamp"] < self.ttl: self.hits += 1 self.cache.move_to_end(chave) return entrada["dados"] else: del self.cache[chave] self.misses += 1 return None def set(self, cpf: str, dados: dict): chave = self._hash(cpf) if len(self.cache) >= self.max_entries: self.cache.popitem(last=False) self.cache[chave] = { "dados": dados, "timestamp": time.time() } def _hash(self, cpf: str) -> str: return hashlib.sha256(cpf.encode()).hexdigest() @property def taxa_hit(self) -> float: total = self.hits + self.misses return (self.hits / total * 100) if total > 0 else 0 class CPFServiceResilient: def __init__(self, api_key: str): self.api_key = api_key self.base_url = "https://api.cpfhub.io/cpf" self.cache = CPFCache(max_entries=10000, ttl_seconds=86400) self.session = requests.Session() self.session.headers["x-api-key"] = api_key def consultar(self, cpf: str) -> dict: cpf_limpo = "".join(c for c in cpf if c.isdigit()) # Camada 1: Cache cached = self.cache.get(cpf_limpo) if cached: return {**cached, "origem": "cache"} # Camada 2: API com timeout try: response = self.session.get( f"{self.base_url}/{cpf_limpo}", timeout=10 ) if response.status_code == 200: dados = response.json() if dados.get("success"): self.cache.set(cpf_limpo, dados["data"]) return {**dados["data"], "origem": "api"} return {"valido": False, "erro": "CPF não encontrado", "origem": "api"} except Exception: # Camada 3: Fallback local return self._validacao_local(cpf_limpo) def _validacao_local(self, cpf: str) -> dict: digits = [int(d) for d in cpf] if len(digits) != 11 or len(set(digits)) == 1: return {"valido": False, "erro": "Formato inválido", "origem": "local"} soma1 = sum(digits[i] * (10 - i) for i in range(9)) check1 = 0 if soma1 % 11 < 2 else 11 - (soma1 % 11) soma2 = sum(digits[i] * (11 - i) for i in range(10)) check2 = 0 if soma2 % 11 < 2 else 11 - (soma2 % 11) if digits[9] == check1 and digits[10] == check2: return { "valido": None, "nota": "Formato válido, dados pendentes de confirmação", "origem": "local" } return {"valido": False, "erro": "CPF inválido", "origem": "local"} ``` | Cenário | Cache hit | Chamada API | Fallback | Resultado | |---------|-----------|-------------|----------|-----------| | CPF recente | Sim | Não | Não | Instantâneo do cache | | CPF novo, API ok | Não | Sim | Não | Dados completos da API | | CPF novo, API fora | Não | Falha | Sim | Validação local apenas | - **Hash do CPF** -- armazenar o hash em vez do CPF em texto claro protege dados em caso de leak do cache - **LRU eviction** -- quando o cache atinge o limite, as entradas mais antigas são removidas - **Taxa de hit** -- monitore essa métrica para dimensionar o tamanho do cache --- ## Implementando circuit breaker O circuit breaker impede que seu sistema fique tentando chamar uma API que está claramente fora do ar. ```python import time from enum import Enum class CircuitState(Enum): CLOSED = "closed" OPEN = "open" HALF_OPEN = "half_open" class CircuitBreaker: def __init__( self, falhas_para_abrir: int = 5, tempo_recuperacao: int = 60, sucesso_para_fechar: int = 3 ): self.estado = CircuitState.CLOSED self.falhas_consecutivas = 0 self.sucessos_half_open = 0 self.falhas_para_abrir = falhas_para_abrir self.tempo_recuperacao = tempo_recuperacao self.sucesso_para_fechar = sucesso_para_fechar self.ultimo_falha = 0 def pode_executar(self) -> bool: if self.estado == CircuitState.CLOSED: return True if self.estado == CircuitState.OPEN: if time.time() - self.ultimo_falha > self.tempo_recuperacao: self.estado = CircuitState.HALF_OPEN self.sucessos_half_open = 0 return True return False return True # HALF_OPEN permite tentativa def registrar_sucesso(self): if self.estado == CircuitState.HALF_OPEN: self.sucessos_half_open += 1 if self.sucessos_half_open >= self.sucesso_para_fechar: self.estado = CircuitState.CLOSED self.falhas_consecutivas = 0 else: self.falhas_consecutivas = 0 def registrar_falha(self): self.falhas_consecutivas += 1 self.ultimo_falha = time.time() if self.falhas_consecutivas >= self.falhas_para_abrir: self.estado = CircuitState.OPEN # Integração com o serviço class CPFServiceComCircuitBreaker: def __init__(self, api_key: str): self.service = CPFServiceResilient(api_key) self.breaker = CircuitBreaker(falhas_para_abrir=5, tempo_recuperacao=60) def consultar(self, cpf: str) -> dict: if not self.breaker.pode_executar(): return self.service._validacao_local( "".join(c for c in cpf if c.isdigit()) ) resultado = self.service.consultar(cpf) if resultado.get("origem") == "api": self.breaker.registrar_sucesso() elif resultado.get("origem") == "local": self.breaker.registrar_falha() return resultado ``` | Estado | Comportamento | Transição | |--------|-------------|-----------| | **Closed** | Todas as requisições passam | Abre após 5 falhas consecutivas | | **Open** | Nenhuma requisição passa, vai direto ao fallback | Semi-abre após 60 segundos | | **Half-Open** | Permite 1 requisição de teste | Fecha após 3 sucessos consecutivos | - **Proteção mútua** -- protege tanto o seu sistema (evitando timeouts) quanto a API (reduzindo carga) - **Recuperação automática** -- o circuito se fecha automaticamente quando a API volta a funcionar - **Feedback imediato** -- quando o circuito está aberto, o fallback responde instantaneamente --- ## Monitorando a saúde da integração Visibilidade sobre o comportamento da integração é essencial para agir proativamente. ```python class MonitorSaude: def __init__(self, service: CPFServiceComCircuitBreaker): self.service = service self.metricas = { "total": 0, "cache_hits": 0, "api_calls": 0, "fallbacks": 0, "erros": 0 } def registrar(self, resultado: dict): self.metricas["total"] += 1 origem = resultado.get("origem", "desconhecido") if origem == "cache": self.metricas["cache_hits"] += 1 elif origem == "api": self.metricas["api_calls"] += 1 elif origem == "local": self.metricas["fallbacks"] += 1 if resultado.get("valido") is False: self.metricas["erros"] += 1 def relatorio(self) -> dict: total = max(self.metricas["total"], 1) return { "total_consultas": self.metricas["total"], "taxa_cache": f"{self.metricas['cache_hits']/total*100:.1f}%", "taxa_api": f"{self.metricas['api_calls']/total*100:.1f}%", "taxa_fallback": f"{self.metricas['fallbacks']/total*100:.1f}%", "circuit_breaker": self.service.breaker.estado.value, "cache_hit_rate": f"{self.service.service.cache.taxa_hit:.1f}%" } ``` - **Taxa de fallback** -- se está alta, a API pode estar instável e merece investigação - **Estado do circuit breaker** -- monitorar se está abrindo com frequência indica problemas - **Cache hit rate** -- taxa acima de 50% indica boa economia de chamadas à API --- ## Perguntas frequentes ### O que é validação local de CPF e quando usá-la como fallback? A validação local verifica matematicamente os dígitos verificadores do CPF sem consultar nenhuma API. Ela confirma que o número tem formato válido, mas não verifica se o CPF existe de fato ou quem é o titular. Use como fallback apenas quando a API estiver indisponível — deixando claro ao usuário que a validação completa será feita posteriormente. ### Qual TTL (tempo de vida) é recomendado para o cache de respostas da API de CPF? Para dados cadastrais de CPF, um TTL de 24 horas é razoável: mudanças de nome ou situação cadastral são raras e o Banco de Dados da Receita Federal é atualizado com baixa frequência. Para contextos onde atualização em tempo real é crítica (como KYC regulatório), reduza o TTL para 1-6 horas ou não faça cache da resposta. ### A API CPFHub.io bloqueia requisições quando o limite do plano gratuito é atingido? Não. O CPFHub.io não bloqueia consultas ao atingir o limite mensal: cada consulta excedente é cobrada automaticamente a R$0,15, sem interrupção. Isso significa que o circuit breaker não precisa ser ativado por razão de limite — ele serve apenas para proteger contra falhas de rede ou indisponibilidade temporária da API. ### Como o [CERT.br](https://www.cert.br/) recomenda tratar dependências externas em sistemas críticos? O CERT.br orienta que sistemas críticos não devem depender de um único ponto de falha externo. Aplicar circuit breaker, timeout agressivo e fallback local são práticas alinhadas com esse princípio — garantindo que uma falha em um fornecedor externo não derrube toda a operação interna. ### Leia também - [SLA de API de CPF: níveis de disponibilidade e o que exigir do seu provedor](https://cpfhub.io/blog/sla-api-cpf-niveis-disponibilidade) - [Como fazer fallback automático quando a API de CPF está fora do ar](https://cpfhub.io/blog/como-fazer-fallback-automatico-quando-a-api-de-cpf-esta-fora-do-ar) - [Quando migrar de API gratuita para versão paga](https://cpfhub.io/blog/quando-migrar-api-gratuita-versao-paga) - [API de CPF grátis para desenvolvedores: como começar em 5 minutos](https://cpfhub.io/blog/api-cpf-gratis-desenvolvedores-comecar-5-minutos) --- ## Conclusão Usar uma API gratuita de CPF não significa aceitar indisponibilidade. Com cache local, circuit breaker, fallback de validação local e monitoramento de saúde, seu sistema mantém funcionalidade mesmo durante períodos de instabilidade da API. Essas estratégias são boas práticas independentemente do plano: elas protegem seu sistema e reduzem custos mesmo em planos pagos. Cadastre-se em [cpfhub.io](https://www.cpfhub.io/) — 50 consultas mensais gratuitas, sem cartão de crédito — e construa sua camada de resiliência já na primeira semana de integração.