Como usar paginação e filtros ao consultar CPFs em lote via API

Para consultar CPFs em lote com eficiência via API, divida a lista em páginas de 20 a 100 registros, aplique filtros de deduplicação e validação sintática antes de cada requisição e grave checkpoints por página para retomar processamentos interrompidos sem perder progresso.

Introdução

Processar grandes volumes de CPFs via API exige mais do que simplesmente enviar requisições em sequência. Sem paginação e filtros adequados, o processamento se torna lento, consume cota desnecessariamente e sobrecarrega tanto o cliente quanto o servidor.

Por que paginação importa em consultas em lote

Processar milhares de CPFs de uma vez cria problemas em cascata:

• Memória -- carregar todos os resultados em memória simultaneamente pode causar out-of-memory
• Timeout -- processar uma lista muito grande pode exceder o tempo limite da conexão
• Recuperação de falhas -- sem paginação, uma falha no meio do processo obriga a recomeçar do zero
• Visibilidade -- sem páginas, é impossível saber o progresso atual do processamento
• Controle de custo -- enviar tudo de uma vez dificulta rastrear o consumo de consultas e planejar o orçamento

Modelos de paginação para consultas de CPF

Existem diferentes estratégias de paginação, cada uma com suas vantagens:

Implementação de processamento paginado com Python

import requests
import time
import math
from typing import List, Optional

class CPFBatchProcessor:
    def __init__(self, api_key: str, page_size: int = 50):
        self.api_key = api_key
        self.page_size = page_size
        self.base_url = "https://api.cpfhub.io/cpf"

    def paginar(self, cpfs: List[str]) -> List[List[str]]:
        """Divide a lista de CPFs em páginas"""
        total_paginas = math.ceil(len(cpfs) / self.page_size)
        paginas = []
        for i in range(total_paginas):
            inicio = i * self.page_size
            fim = inicio + self.page_size
            paginas.append(cpfs[inicio:fim])
        return paginas

    def filtrar_validos(self, cpfs: List[str]) -> List[str]:
        """Filtra CPFs sintaticamente válidos antes de consultar a API"""
        validos = []
        for cpf in cpfs:
            cpf_limpo = cpf.replace(".", "").replace("-", "").strip()
            if len(cpf_limpo) == 11 and cpf_limpo.isdigit():
                if cpf_limpo != cpf_limpo[0] * 11:
                    validos.append(cpf_limpo)
        return validos

    def consultar(self, cpf: str) -> Optional[dict]:
        try:
            response = requests.get(
                f"{self.base_url}/{cpf}",
                headers={"x-api-key": self.api_key},
                timeout=10
            )
            if response.status_code == 200 and response.json()["success"]:
                d = response.json()["data"]
                return {
                    "cpf": d["cpf"],
                    "nome": d["name"],
                    "genero": d["gender"],
                    "nascimento": d["birthDate"],
                    "ano": d["year"]
                }
        except Exception:
            pass
        return None

    def processar_lote(self, cpfs_brutos: List[str],
                    filtro_genero: str = None,
                    filtro_ano_min: int = None):
        # Etapa 1: Filtrar e limpar
        cpfs = self.filtrar_validos(cpfs_brutos)
        removidos = len(cpfs_brutos) - len(cpfs)
        print(f"CPFs válidos: {len(cpfs)} (removidos: {removidos})")

        # Etapa 2: Paginar
        paginas = self.paginar(cpfs)
        print(f"Total de páginas: {len(paginas)} ({self.page_size} por página)")

        resultados = []

        # Etapa 3: Processar página a página
        for num_pag, pagina in enumerate(paginas, 1):
            print(f"\nProcessando página {num_pag}/{len(paginas)}...")

            for cpf in pagina:
                resultado = self.consultar(cpf)

                if resultado:
                    # Aplicar filtros pós-consulta
                    if filtro_genero and resultado["genero"] != filtro_genero:
                        continue
                    if filtro_ano_min and resultado["ano"] < filtro_ano_min:
                        continue
                    resultados.append(resultado)

            time.sleep(0.5)

            print(f" Resultados acumulados: {len(resultados)}")

        return resultados

# Uso com filtros
processor = CPFBatchProcessor("SUA_CHAVE_AQUI", page_size=20)
resultados = processor.processar_lote(
    cpfs_brutos=["123.456.789-09", "987.654.321-00", "111.222.333-44"],
    filtro_genero="M",
    filtro_ano_min=1980
)
print(f"\nTotal de resultados filtrados: {len(resultados)}")

Filtros úteis para processamento em lote

Aplique filtros para reduzir o volume de dados processados e focar no que importa:

• Filtro por gênero -- processar apenas CPFs de um gênero específico quando a operação exige segmentação
• Filtro por faixa etária -- limitar consultas a pessoas dentro de uma faixa de idade relevante para o negócio
• Filtro de deduplicação -- remover CPFs duplicados antes do processamento para economizar cota
• Filtro de cache -- excluir CPFs que já foram consultados recentemente e estão em cache
• Filtro sintático -- eliminar CPFs com formato inválido antes de consumir a API

Otimizações avançadas

Para volumes de centenas de milhares de CPFs, aplique otimizações adicionais:

• Processamento paralelo com pool -- utilize múltiplos workers que compartilham o controle de consumo global
• Checkpoint por página -- salve o número da última página processada para retomada em caso de falha
• Compressão de resultados -- armazene resultados em formato compactado para economizar espaço em disco
• Streaming de resultados -- escreva resultados em disco conforme são processados, sem acumular em memória
• Priorização inteligente -- ordene os CPFs por probabilidade de sucesso, processando os mais prováveis primeiro

O CERT.br recomenda que sistemas que processam dados pessoais em lote implementem registro de auditoria por lote, garantindo rastreabilidade em caso de incidentes.

Perguntas frequentes

Como calcular quantas consultas em lote cabem no plano gratuito da CPFHub.io?

O plano gratuito oferece 50 consultas mensais. Para um lote típico de 50 CPFs com página de 10, isso ocupa toda a cota mensal em uma única execução. Para volumes maiores, o plano Pro a R$149/mês inclui 1.000 consultas; ao ultrapassar esse limite, cada consulta adicional custa R$0,15 — a API nunca bloqueia o acesso, apenas registra o excedente na fatura.

Qual tamanho de página é mais indicado para processamento em lote de CPFs?

Páginas de 20 a 50 registros oferecem bom equilíbrio entre rastreabilidade e eficiência. Páginas menores facilitam a retomada em caso de falha; páginas maiores reduzem o overhead de controle. Para lotes com priorização crítica, prefira páginas de 10 registros com checkpoint após cada uma.

É possível aplicar filtros antes de consumir a API para economizar consultas?

Sim. Filtros sintáticos (CPF com 11 dígitos, não sequência de dígitos iguais) e filtros de deduplicação devem ser aplicados localmente antes de qualquer chamada. Isso pode reduzir o volume efetivo em 5% a 20% dependendo da qualidade da lista original, sem consumir cota da API.

Como retomar um processamento em lote interrompido sem reprocessar CPFs já consultados?

Salve o índice da última página concluída em um arquivo de checkpoint após cada página. Ao retomar, carregue o checkpoint e pule as páginas já processadas. Combine isso com um arquivo de resultados parciais em formato JSONL (uma linha por registro) para garantir que nada seja perdido ou duplicado.

Conclusão

Paginação e filtros são técnicas essenciais para consultar CPFs em lote de forma eficiente e controlada. Ao dividir o processamento em páginas gerenciáveis, filtrar dados desnecessários e implementar checkpoints de progresso, você transforma uma operação caótica em um processo previsível e auditável.

Cadastre-se em cpfhub.io — 50 consultas mensais gratuitas, sem cartão de crédito — e processe seus primeiros lotes de CPF com controle total de paginação e filtros.