Load balancing entre múltiplas chaves de API de CPF

Quando o volume de consultas de CPF ultrapassa a cota de uma única chave de API, ou quando a operação exige redundância para garantir disponibilidade, a solução é distribuir as requisições entre múltiplas chaves. Essa estratégia — conhecida como load balancing no nível da aplicação — maximiza a cota efetiva, adiciona uma camada extra de resiliência e permite segregar consumo por ambiente ou departamento sem depender de um único ponto de falha.

Quando usar múltiplas chaves

Cenários típicos

Volume alto distribuído entre departamentos: diferentes áreas da empresa têm suas próprias cotas e orçamentos.
Redundância: se uma chave for comprometida ou atingir o limite, outra assume automaticamente.
Ambientes separados: chaves distintas para desenvolvimento, staging e produção.
Escala além do plano individual: combinar múltiplas chaves Pro para atingir volume corporativo enquanto negocia um plano Corporativo.

Considerações importantes

Antes de implementar, verifique os termos de uso do provedor. A CPFHub.io

Estratégia 1 -- Round-robin

A estratégia mais simples: distribui as requisições de forma circular entre as chaves disponíveis.

import requests
import logging
from itertools import cycle
from threading import Lock
from typing import Optional, Dict, List

logger = logging.getLogger(__name__)

class RoundRobinBalancer:
    """Distribui consultas de CPF em round-robin entre múltiplas chaves."""

    BASE_URL = "https://api.cpfhub.io/cpf"
    TIMEOUT = 30

    def __init__(self, api_keys: List[str]):
    if not api_keys:
    raise ValueError("Pelo menos uma chave de API e necessaria")
    self._keys = api_keys
    self._cycle = cycle(range(len(api_keys)))
    self._lock = Lock()

    def _proxima_chave(self) -> str:
    with self._lock:
    idx = next(self._cycle)
    return self._keys[idx]

    def consultar(self, cpf: str) -> Optional[Dict]:
    chave = self._proxima_chave()
    cpf_limpo = cpf.replace(".", "").replace("-", "")

    try:
    response = requests.get(
    f"{self.BASE_URL}/{cpf_limpo}",
    headers={
    "x-api-key": chave,
    "Accept": "application/json"
    },
    timeout=self.TIMEOUT
    )
    response.raise_for_status()
    return response.json()

    except requests.exceptions.Timeout:
    logger.error("Timeout na consulta de CPF")
    return None
    except requests.exceptions.RequestException as e:
    logger.error(f"Erro: {e}")
    return None

# Uso
balancer = RoundRobinBalancer(api_keys=[
    "chave_pro_1",
    "chave_pro_2",
    "chave_pro_3"
])

resultado = balancer.consultar("12345678900")

Vantagens

Implementação simples.
Distribuição uniforme de carga.

Desvantagens

Não considera o consumo atual de cada chave.
Se uma chave está com problema, ainda recebe requisições.

Estratégia 2 -- Baseada em consumo

Direciona as requisições para a chave com mais cota disponível:

from dataclasses import dataclass, field
from datetime import date

@dataclass
class ChaveAPI:
    """Representa uma chave de API com controle de consumo."""
    key: str
    cota_mensal: int
    consumo_diario: Dict[str, int] = field(default_factory=dict)

    @property
    def consumo_mes_atual(self) -> int:
    mes = date.today().strftime("%Y-%m")
    return sum(v for k, v in self.consumo_diario.items() if k.startswith(mes))

    @property
    def cota_disponivel(self) -> int:
    return max(0, self.cota_mensal - self.consumo_mes_atual)

    def registrar_uso(self):
    hoje = date.today().isoformat()
    self.consumo_diario[hoje] = self.consumo_diario.get(hoje, 0) + 1

class ConsumoBalancer:
    """Distribui consultas priorizando a chave com mais cota disponível."""

    BASE_URL = "https://api.cpfhub.io/cpf"
    TIMEOUT = 30

    def __init__(self, chaves: List[ChaveAPI]):
    self._chaves = chaves
    self._lock = Lock()

    def _selecionar_chave(self) -> Optional[ChaveAPI]:
    with self._lock:
    disponiveis = [c for c in self._chaves if c.cota_disponivel > 0]
    if not disponiveis:
    return None
    return max(disponiveis, key=lambda c: c.cota_disponivel)

    def consultar(self, cpf: str) -> Optional[Dict]:
    chave = self._selecionar_chave()
    if not chave:
    logger.error("Nenhuma chave com cota disponivel")
    return None

    cpf_limpo = cpf.replace(".", "").replace("-", "")

    try:
    response = requests.get(
    f"{self.BASE_URL}/{cpf_limpo}",
    headers={
    "x-api-key": chave.key,
    "Accept": "application/json"
    },
    timeout=self.TIMEOUT
    )
    response.raise_for_status()
    chave.registrar_uso()
    return response.json()

    except requests.exceptions.Timeout:
    logger.error("Timeout na consulta")
    return None
    except requests.exceptions.RequestException as e:
    logger.error(f"Erro: {e}")
    return None

    def status(self) -> str:
    """Retorna status de consumo de todas as chaves."""
    linhas = ["=== Status das Chaves ==="]
    for i, chave in enumerate(self._chaves):
    linhas.append(
    f"Chave {i+1}: {chave.consumo_mes_atual}/{chave.cota_mensal} "
    f"({chave.cota_disponivel} restantes)"
    )
    return "\n".join(linhas)

# Uso
chaves = [
    ChaveAPI(key="chave_pro_1", cota_mensal=1000),
    ChaveAPI(key="chave_pro_2", cota_mensal=1000),
]

balancer = ConsumoBalancer(chaves=chaves)
resultado = balancer.consultar("12345678900")
print(balancer.status())

Estratégia 3 -- Com fallback automático

Se a chave primária falha, automaticamente tenta a próxima. Note que a CPFHub.io não bloqueia ao atingir a cota inclusa — a API continua respondendo e cobra R$0,15 por consulta excedente. O fallback aqui é útil para erros de rede ou respostas 5xx:

class FallbackBalancer:
    """Tenta chaves em sequência até obter sucesso."""

    BASE_URL = "https://api.cpfhub.io/cpf"
    TIMEOUT = 30

    def __init__(self, api_keys: List[str]):
    self._keys = api_keys

    def consultar(self, cpf: str) -> Optional[Dict]:
    cpf_limpo = cpf.replace(".", "").replace("-", "")

    for i, chave in enumerate(self._keys):
    try:
    response = requests.get(
    f"{self.BASE_URL}/{cpf_limpo}",
    headers={
    "x-api-key": chave,
    "Accept": "application/json"
    },
    timeout=self.TIMEOUT
    )

    if response.status_code == 200:
    return response.json()

    if response.status_code >= 500:
    logger.warning(f"Erro {response.status_code} na chave {i+1}. Tentando proxima...")
    continue

    # Erros 4xx nao justificam fallback entre chaves
    response.raise_for_status()

    except requests.exceptions.Timeout:
    logger.warning(f"Timeout na chave {i+1}. Tentando proxima...")
    continue
    except requests.exceptions.ConnectionError:
    logger.warning(f"Erro de conexao na chave {i+1}. Tentando proxima...")
    continue

    logger.error("Todas as chaves falharam")
    return None

Implementação em Node.js

const axios = require("axios");

const BASE_URL = "https://api.cpfhub.io/cpf";
const TIMEOUT_MS = 30000;

class LoadBalancer {
    constructor(apiKeys) {
    this.apiKeys = apiKeys;
    this.currentIndex = 0;
    }

    nextKey() {
    const key = this.apiKeys[this.currentIndex];
    this.currentIndex = (this.currentIndex + 1) % this.apiKeys.length;
    return key;
    }

    async consultar(cpf) {
    const cpfLimpo = cpf.replace(/\D/g, "");

    // Tenta todas as chaves como fallback
    for (let i = 0; i < this.apiKeys.length; i++) {
    const key = this.nextKey();

    try {
    const response = await axios.get(`${BASE_URL}/${cpfLimpo}`, {
    headers: {
    "x-api-key": key,
    "Accept": "application/json"
    },
    timeout: TIMEOUT_MS
    });

    return response.data;

    } catch (error) {
    const status = error.response ? error.response.status : null;
    if (status >= 500 || !status) {
    console.warn(`Chave ${i + 1} falhou (${status || "timeout"}). Tentando proxima...`);
    continue;
    }
    throw error; // Erros nao-retentaveis
    }
    }

    return null;
    }
}

// Uso
const balancer = new LoadBalancer([
    "chave_pro_1",
    "chave_pro_2",
    "chave_pro_3"
]);

balancer.consultar("12345678900").then(result => {
    if (result && result.success) {
    console.log(`Nome: ${result.data.name}`);
    }
});

Verificando o status via cURL

Para monitorar a saúde de cada chave individualmente:

# Testar chave 1
curl -s -o /dev/null -w "Chave 1: HTTP %{http_code} em %{time_total}s\n" \
    "https://api.cpfhub.io/cpf/12345678900" \
    -H "x-api-key: CHAVE_1" \
    -H "Accept: application/json" \
    --connect-timeout 10 \
    --max-time 30

# Testar chave 2
curl -s -o /dev/null -w "Chave 2: HTTP %{http_code} em %{time_total}s\n" \
    "https://api.cpfhub.io/cpf/12345678900" \
    -H "x-api-key: CHAVE_2" \
    -H "Accept: application/json" \
    --connect-timeout 10 \
    --max-time 30

Comparativo das estratégias

Aspecto	Round-robin	Baseada em consumo	Com fallback
Complexidade	Baixa	Média	Baixa
Distribuição	Uniforme	Inteligente	Sequencial
Resiliência	Baixa	Média	Alta
Controle de cota	Nenhum	Completo	Parcial (via erros 5xx)
Melhor para	Volume previsível	Gestão de custos	Alta disponibilidade

Quando preferir o plano Corporativo

Se você está gerenciando mais de duas chaves Pro para atingir o volume necessário, provavelmente é hora de considerar o plano Corporativo da CPFHub.io

Perguntas frequentes

O que é load balancing de chaves de API e por que usá-lo em consultas de CPF?

Load balancing de chaves de API é a técnica de distribuir requisições entre múltiplas credenciais de autenticação, em vez de concentrar tudo em uma única chave. Para consultas de CPF em alto volume, isso aumenta a cota efetiva, adiciona redundância e permite segregar consumo por ambiente ou departamento. A estratégia mais simples é o round-robin; a mais resiliente combina seleção por cota disponível com fallback automático para erros de rede ou instabilidades de infraestrutura.

A CPFHub.io bloqueia quando a cota mensal é atingida?

Não. A CPFHub.io não bloqueia nem retorna erro ao atingir a cota inclusa no plano. O plano gratuito inclui 50 consultas/mês e o Pro inclui 1.000 por R$149/mês. Após o limite, a API continua respondendo normalmente e cobra R$0,15 por consulta excedente — o que simplifica a lógica de fallback, pois não há código de erro específico de cota para tratar.

Como monitorar qual chave está consumindo mais cota?

A abordagem mais robusta é manter um contador local por chave, incrementado a cada consulta bem-sucedida. Para múltiplas instâncias da aplicação, use um contador centralizado em Redis com incremento atômico (INCR) e expiração automática no início de cada mês, eliminando a necessidade de lógica de reset manual.

Quantas chaves Pro são necessárias para substituir um plano Corporativo?

Depende do volume. Cada chave Pro oferece 1.000 consultas mensais incluídas, a R$149/mês. Para 5.000 consultas/mês seriam 5 chaves (R$745/mês). A partir de 3 ou 4 chaves gerenciadas simultaneamente, vale avaliar o plano Corporativo da CPFHub.io pela simplificação operacional e pelo suporte dedicado.

Conclusão

O load balancing entre múltiplas chaves de API é uma técnica poderosa para escalar consultas de CPF, adicionar redundância e gerenciar cotas de forma inteligente. A estratégia ideal depende do seu cenário: round-robin para simplicidade, baseada em consumo para controle de custos e fallback para máxima disponibilidade.

Para operações de grande escala, a CPFHub.io

Cadastre-se em cpfhub.io

CPFHub.io

Pronto para integrar a API?

50 consultas gratuitas para testar agora. Sem cartão de crédito. Acesso imediato à documentação.

Começar grátis Ver documentação

Sobre a redação

Redação CPFHub.io

Time editorial especializado em APIs de CPF, identidade digital e compliance no mercado brasileiro. Produzimos guias técnicos, análises regulatórias e tutoriais sobre LGPD e KYC para desenvolvedores e líderes de produto.

API de CPF: como fazer load balancing entre múltiplas chaves de API