Paginacao para consultas de CPF em massa via API

Para processar listas inteiras de CPFs via API, a CPFHub.io recomenda dividir o volume em lotes controlados com intervalos entre requisições, salvando checkpoints para retomar em caso de interrupção. Essa abordagem garante processamento eficiente, custo previsível e rastreabilidade completa — sem perder dados nem comprometer a estabilidade da integração.

Introdução

Muitas empresas precisam validar listas inteiras de CPFs -- seja para atualizar cadastros, verificar beneficiarios ou enriquecer bases de dados. Como a API da CPFHub.io processa um CPF por requisição, consultas em massa exigem uma estrategia de paginacao e processamento em lote que respeite os rate limits sem comprometer a eficiência.

Entendendo os limites e custos por plano

Antes de processar em lote, e essencial conhecer os limites de cada plano:

Plano	Intervalo recomendado	Consultas/mês	Tempo por lote de 100 CPFs
Gratuito	1 req a cada 2 seg	50 + R$0,15/extra	~200 segundos
Pro	1 req por segundo	1.000 + R$0,15/extra	~100 segundos
Corporativo	Personalizado	Personalizado	Sob consulta

A API da CPFHub.io nunca bloqueia requisições ao atingir a cota mensal — cada consulta excedente é cobrada automaticamente a R$0,15. Por isso, o planejamento do volume é importante principalmente para o controle de custos, não para evitar interrupções.

Estrategia 1: Processamento sequencial com intervalo

A abordagem mais simples -- processar um CPF por vez com pausa entre chamadas:

import requests
import time
import csv

API_KEY = 'SUA_CHAVE_DE_API'
INTERVALO = 2 # segundos entre chamadas (plano Gratuito)

headers = {
    'x-api-key': API_KEY,
    'Accept': 'application/json'
}

def processar_lista_cpfs(arquivo_entrada, arquivo_saida):
    with open(arquivo_entrada, 'r') as f_in, \
    open(arquivo_saida, 'w', newline='') as f_out:

    reader = csv.reader(f_in)
    writer = csv.writer(f_out)
    writer.writerow(['cpf', 'nome', 'genero', 'nascimento', 'status'])

    for i, row in enumerate(reader):
    cpf = row[0].strip().replace('.', '').replace('-', '')

    try:
    response = requests.get(
    f'https://api.cpfhub.io/cpf/{cpf}',
    headers=headers,
    timeout=15
    )

    if response.status_code == 200:
    data = response.json()
    if data.get('success'):
    d = data['data']
    writer.writerow([
    d['cpf'], d['name'], d['gender'],
    d['birthDate'], 'encontrado'
    ])
    else:
    writer.writerow([cpf, '', '', '', 'nao_encontrado'])
    else:
    writer.writerow([cpf, '', '', '', f'erro_{response.status_code}'])

    except requests.exceptions.Timeout:
    writer.writerow([cpf, '', '', '', 'timeout'])

    if (i + 1) % 10 == 0:
    print(f'Processados: {i + 1}')

    time.sleep(INTERVALO)

processar_lista_cpfs('cpfs_entrada.csv', 'cpfs_resultado.csv')

Estrategia 2: Processamento em lotes com checkpoint

Para listas grandes, divida em lotes e salve o progresso para retomar em caso de interrupcao:

import requests
import time
import json
import os

API_KEY = 'SUA_CHAVE_DE_API'
TAMANHO_LOTE = 50
INTERVALO = 2
CHECKPOINT_FILE = 'checkpoint.json'

headers = {
    'x-api-key': API_KEY,
    'Accept': 'application/json'
}

def carregar_checkpoint():
    if os.path.exists(CHECKPOINT_FILE):
    with open(CHECKPOINT_FILE, 'r') as f:
    return json.load(f)
    return {'ultimo_indice': 0, 'resultados': []}

def salvar_checkpoint(checkpoint):
    with open(CHECKPOINT_FILE, 'w') as f:
    json.dump(checkpoint, f)

def processar_em_lotes(lista_cpfs):
    checkpoint = carregar_checkpoint()
    inicio = checkpoint['ultimo_indice']
    resultados = checkpoint['resultados']

    total = len(lista_cpfs)
    print(f'Retomando a partir do indice {inicio} de {total}')

    for i in range(inicio, total):
    cpf = lista_cpfs[i]

    try:
    response = requests.get(
    f'https://api.cpfhub.io/cpf/{cpf}',
    headers=headers,
    timeout=15
    )
    resultados.append({
    'cpf': cpf,
    'status': response.status_code,
    'data': response.json() if response.status_code == 200 else None
    })
    except Exception as e:
    resultados.append({
    'cpf': cpf,
    'status': 'erro',
    'data': str(e)
    })

    # Salvar checkpoint a cada lote
    if (i + 1) % TAMANHO_LOTE == 0:
    checkpoint = {'ultimo_indice': i + 1, 'resultados': resultados}
    salvar_checkpoint(checkpoint)
    print(f'Checkpoint salvo: {i + 1}/{total}')

    time.sleep(INTERVALO)

    # Salvar resultado final
    salvar_checkpoint({'ultimo_indice': total, 'resultados': resultados})
    return resultados

Estrategia 3: Processamento concorrente controlado

Para o plano Pro (1 req/segundo), você pode usar um semaforo para controlar a concorrencia:

import asyncio
import aiohttp

API_KEY = 'SUA_CHAVE_DE_API'
MAX_CONCURRENT = 1 # respeitar rate limit
INTERVALO = 1.0

semaforo = asyncio.Semaphore(MAX_CONCURRENT)

async def consultar_cpf(session, cpf):
    async with semaforo:
    url = f'https://api.cpfhub.io/cpf/{cpf}'
    headers = {
    'x-api-key': API_KEY,
    'Accept': 'application/json'
    }
    try:
    async with session.get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=15)) as response:
    data = await response.json()
    await asyncio.sleep(INTERVALO)
    return {'cpf': cpf, 'status': response.status, 'data': data}
    except Exception as e:
    await asyncio.sleep(INTERVALO)
    return {'cpf': cpf, 'status': 'erro', 'data': str(e)}

async def processar_lote(lista_cpfs):
    async with aiohttp.ClientSession() as session:
    tarefas = [consultar_cpf(session, cpf) for cpf in lista_cpfs]
    resultados = await asyncio.gather(*tarefas)
    return resultados

# Executar
lista = ['12345678900', '98765432100', '11122233344']
resultados = asyncio.run(processar_lote(lista))

Organizando a fila de entrada

Remover duplicatas

lista_cpfs = list(set(lista_cpfs_bruta))
print(f'CPFs unicos: {len(lista_cpfs)} (de {len(lista_cpfs_bruta)})')

Validar formato antes de consultar

import re

def cpf_formato_valido(cpf):
    cpf_limpo = re.sub(r'\D', '', cpf)
    return len(cpf_limpo) == 11 and cpf_limpo.isdigit()

lista_valida = [cpf for cpf in lista_cpfs if cpf_formato_valido(cpf)]
lista_invalida = [cpf for cpf in lista_cpfs if not cpf_formato_valido(cpf)]

print(f'Validos para consulta: {len(lista_valida)}')
print(f'Formato invalido (pular): {len(lista_invalida)}')

Estimativa de tempo e custo

Antes de iniciar o processamento, calcule o tempo estimado:

def estimar_processamento(total_cpfs, intervalo_seg, cota_mensal, preco_excedente=0.15):
    tempo_total_seg = total_cpfs * intervalo_seg
    tempo_horas = tempo_total_seg / 3600

    dentro_cota = min(total_cpfs, cota_mensal)
    excedente = max(0, total_cpfs - cota_mensal)
    custo_excedente = excedente * preco_excedente

    print(f'Total de CPFs: {total_cpfs}')
    print(f'Tempo estimado: {tempo_horas:.1f} horas')
    print(f'Consultas dentro da cota: {dentro_cota}')
    print(f'Consultas excedentes: {excedente}')
    print(f'Custo excedente: R$ {custo_excedente:.2f}')

# Plano Pro
estimar_processamento(5000, 1, 1000)

Tratamento de erros e retentativas

Status 500/503 -- Retentativa com backoff exponencial (5s, 15s, 45s).
Timeout -- Adicionar a uma fila de retentativa para processar no final.
Erro de rede -- Verificar conectividade e retomar do checkpoint.
Custo excedente -- Acompanhe o volume consultado no painel app.cpfhub.io/settings/billing. A API cobra R$0,15 por consulta acima da cota sem interromper o processamento, então o controle proativo do volume é a melhor estratégia de custo.

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 1 segundo, 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

Processar consultas de CPF em massa exige planejamento para controlar custos, manter o progresso salvo e tratar erros adequadamente. Com checkpoints, validação previa e controle de concorrencia, você processa grandes volumes de forma eficiente e previsível — sabendo exatamente quantas consultas foram realizadas e qual o custo final.

Cadastre-se em cpfhub.io — 50 consultas mensais gratuitas, sem cartão de crédito — e comece hoje mesmo.

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.

Estrategias de paginacao para consultas de CPF em massa via API