Como Implementar Fallback entre Múltiplos Provedores de API de CPF

Introdução

Depender de um único provedor de API de CPF é um ponto único de falha. Quando esse provedor fica indisponível, sua aplicação inteira para de funcionar. A solução é implementar um mecanismo de fallback que automaticamente redireciona as requisições para um provedor alternativo quando o principal falha.

Estratégias de fallback

Existem diferentes estratégias para implementar fallback, cada uma com seus trade-offs. A escolha depende dos requisitos de latência, custo e complexidade da sua aplicação.

• Fallback sequencial -- tenta o provedor principal primeiro e, se falhar, passa para o próximo na lista
• Fallback paralelo -- dispara requisições para todos os provedores simultaneamente e usa a primeira resposta válida
• Circuit breaker -- monitora falhas e "abre" o circuito para provedores instáveis, evitando cascata de erros

Implementando o circuit breaker

O circuit breaker é o coração do sistema de fallback. Ele monitora a taxa de falhas de cada provedor e decide quando parar de enviar requisições para provedores problemáticos.

import time
from enum import Enum
from dataclasses import dataclass, field

class CircuitState(Enum):
    CLOSED = "closed" # Funcionando normalmente
    OPEN = "open" # Bloqueado, provedor com problemas
    HALF_OPEN = "half_open" # Testando se o provedor se recuperou

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout: int = 30
    _failure_count: int = field(default=0, init=False)
    _last_failure_time: float = field(default=0, init=False)
    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)

    @property
    def state(self) -> CircuitState:
    if self._state == CircuitState.OPEN:
    if time.time() - self._last_failure_time > self.recovery_timeout:
    self._state = CircuitState.HALF_OPEN
    return self._state

    def record_success(self):
    self._failure_count = 0
    self._state = CircuitState.CLOSED

    def record_failure(self):
    self._failure_count += 1
    self._last_failure_time = time.time()
    if self._failure_count >= self.failure_threshold:
    self._state = CircuitState.OPEN

    def can_execute(self) -> bool:
    return self.state != CircuitState.OPEN

• CLOSED -- o circuito está fechado e as requisições fluem normalmente para o provedor
• OPEN -- o circuito está aberto após muitas falhas, bloqueando requisições por um período
• HALF_OPEN -- após o período de recuperação, permite uma requisição de teste para verificar se o provedor voltou

Construindo o gerenciador de fallback

O gerenciador combina múltiplos provedores com seus respectivos circuit breakers para criar um sistema de fallback completo.

import requests
import logging

logger = logging.getLogger(__name__)

class CpfProvider:
    def __init__(self, name, url, headers, parser):
    self.name = name
    self.url = url
    self.headers = headers
    self.parser = parser
    self.circuit = CircuitBreaker(failure_threshold=3, recovery_timeout=60)

    def consultar(self, cpf):
    response = requests.get(
    self.url.format(cpf=cpf),
    headers=self.headers,
    timeout=10
    )
    response.raise_for_status()
    return self.parser(response.json())

class FallbackManager:
    def __init__(self, providers: list):
    self.providers = providers

    def consultar(self, cpf: str) -> dict:
    errors = []
    for provider in self.providers:
    if not provider.circuit.can_execute():
    logger.warning(f"Provedor {provider.name}: circuito aberto")
    continue

    try:
    result = provider.consultar(cpf)
    provider.circuit.record_success()
    logger.info(f"Consulta via {provider.name}: sucesso")
    return result
    except Exception as e:
    provider.circuit.record_failure()
    errors.append((provider.name, str(e)))
    logger.error(f"Provedor {provider.name} falhou: {e}")

    raise Exception(f"Todos os provedores falharam: {errors}")

O gerenciador percorre a lista de provedores na ordem de prioridade, pulando aqueles com circuito aberto e registrando sucesso ou falha em cada tentativa.

Configurando os provedores

Com a arquitetura pronta, basta configurar os provedores e seus parsers para normalizar as respostas.

def parser_cpfhub(data):
    return {
    "cpf": data["data"]["cpf"],
    "nome": data["data"]["name"],
    "genero": data["data"]["gender"],
    "nascimento": data["data"]["birthDate"],
    }

def parser_alternativo(data):
    return {
    "cpf": data["documento"],
    "nome": data["nome_completo"],
    "genero": data["sexo"],
    "nascimento": data["data_nascimento"],
    }

# Configuração com prioridade
provedores = [
    CpfProvider(
    name="CPFHub",
    url="https://api.cpfhub.io/cpf/{cpf}",
    headers={"x-api-key": "sua-chave-aqui"},
    parser=parser_cpfhub,
    ),
    CpfProvider(
    name="ProvedorB",
    url="https://provedor-b.com/api/cpf/{cpf}",
    headers={"Authorization": "Bearer token-b"},
    parser=parser_alternativo,
    ),
]

manager = FallbackManager(provedores)

# Uso transparente
resultado = manager.consultar("12345678909")
print(resultado["nome"])

Perguntas frequentes

O que é necessário para implementar validação de CPF neste contexto?

A validação de CPF exige uma chamada à API com o número do documento e a chave de autenticação. A CPFHub.io retorna o status do CPF, nome do titular e data de nascimento em menos de 200ms, permitindo a verificação em tempo real durante o cadastro ou transação.

A API CPFHub.io funciona para todos os volumes de consulta?

Sim. O plano gratuito oferece 50 consultas por mês sem cartão de crédito — ideal para testes e projetos pequenos. Para volumes maiores, o plano Pro inclui 1.000 consultas mensais por R$149. Se o limite for ultrapassado, a API não bloqueia: cobra R$0,15 por consulta adicional.

Como garantir conformidade com a LGPD ao usar uma API de CPF?

Use o CPF apenas para a finalidade declarada ao titular, armazene apenas o necessário (não guarde o CPF cru se um token bastar), implemente controle de acesso aos logs de consulta 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.

Quanto tempo leva para integrar a API CPFHub.io?

A integração básica leva menos de 30 minutos: crie uma conta em cpfhub.io, gere a API key no painel e faça uma chamada GET para https://api.cpfhub.io/cpf/{CPF} com o header x-api-key. A documentação inclui exemplos em Python, Node.js, PHP, Java e outras linguagens.

Conclusão

Implementar fallback entre múltiplos provedores transforma sua integração de um ponto único de falha em um sistema resiliente. Com circuit breakers monitorando a saúde de cada provedor e roteamento automático, sua aplicação mantém a disponibilidade mesmo quando um provedor cai. O plano gratuito da cpfhub.io cobre os testes iniciais sem custo.