Para consumir a API de CPF do CPFHub.io em Python com FastAPI, crie um microserviço que usa httpx para chamadas assíncronas ao endpoint GET https://api.cpfhub.io/cpf/{CPF} com o header x-api-key, valide os dígitos localmente com Pydantic antes de chamar a API, e exponha um endpoint /validar-cpf que retorna nome, data de nascimento e gênero do titular. A integração completa leva menos de 30 minutos.
Introdução
O FastAPI é um dos frameworks Python mais populares para construção de APIs modernas, com tipagem forte, validação automática via Pydantic e suporte nativo a operações assíncronas. Combiná-lo com uma API de consulta de CPF cria um microserviço poderoso para validação de identidade, que pode ser integrado a qualquer aplicação.
Pré-requisitos
-
Python 3.9+ -- FastAPI requer Python 3.7 ou superior, mas recomendamos 3.9+.
-
pip -- Gerenciador de pacotes do Python.
-
Conta no CPFHub.io -- Crie uma conta gratuita em cpfhub.io
Instalação das dependências
pip install fastapi uvicorn httpx pydantic python-dotenv
- fastapi -- O framework web.
- uvicorn -- Servidor ASGI para rodar o FastAPI.
- httpx -- Cliente HTTP assíncrono (alternativa ao requests para código async).
- pydantic -- Validação de dados (já incluso no FastAPI, mas importante mencionar).
- python-dotenv -- Carregar variáveis de ambiente.
Estrutura do projeto
cpf-service/
.env
main.py
models.py
services.py
validators.py
Arquivo de configuração (.env)
CPFHUB_API_KEY=SUA_CHAVE_DE_API
CPFHUB_BASE_URL=https://api.cpfhub.io/cpf
Modelos Pydantic (models.py)
Os modelos Pydantic definem a estrutura dos dados de entrada e saída, com validação automática.
from pydantic import BaseModel, field_validator
import re
class ConsultaCPFRequest(BaseModel):
cpf: str
@field_validator("cpf")
@classmethod
def validar_formato_cpf(cls, v):
cpf_limpo = re.sub(r"\D", "", v)
if len(cpf_limpo) != 11:
raise ValueError("CPF deve ter 11 dígitos")
if cpf_limpo == cpf_limpo[0] * 11:
raise ValueError("CPF com todos os dígitos iguais é inválido")
return cpf_limpo
class DadosCPF(BaseModel):
cpf: str
name: str
nameUpper: str
gender: str
birthDate: str
day: int
month: int
year: int
class ConsultaCPFResponse(BaseModel):
success: bool
data: DadosCPF | None = None
erro: str | None = None
class ValidacaoCPFResponse(BaseModel):
cpf_valido: bool
nome: str | None = None
nascimento: str | None = None
genero: str | None = None
mensagem: str
Serviço de consulta (services.py)
O serviço encapsula a lógica de comunicação com a API do CPFHub.io usando httpx para requisições assíncronas.
import httpx
import os
from dotenv import load_dotenv
from models import ConsultaCPFResponse
load_dotenv()
CPFHUB_API_KEY = os.getenv("CPFHUB_API_KEY")
CPFHUB_BASE_URL = os.getenv("CPFHUB_BASE_URL", "https://api.cpfhub.io/cpf")
class CPFHubService:
def __init__(self):
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(10.0),
headers={
"x-api-key": CPFHUB_API_KEY,
"Accept": "application/json"
}
)
async def consultar_cpf(self, cpf: str) -> ConsultaCPFResponse:
try:
response = await self.client.get(f"{CPFHUB_BASE_URL}/{cpf}")
if response.status_code == 200:
dados = response.json()
return ConsultaCPFResponse(**dados)
elif response.status_code == 401:
return ConsultaCPFResponse(
success=False,
erro="Chave de API inválida."
)
else:
return ConsultaCPFResponse(
success=False,
erro=f"Erro na consulta: HTTP {response.status_code}"
)
except httpx.TimeoutException:
return ConsultaCPFResponse(
success=False,
erro="Timeout na consulta ao CPFHub.io"
)
except httpx.RequestError as e:
return ConsultaCPFResponse(
success=False,
erro=f"Erro de conexão: {str(e)}"
)
async def fechar(self):
await self.client.aclose()
cpfhub_service = CPFHubService()
Validador de dígitos verificadores (validators.py)
def validar_digitos_cpf(cpf: str) -> bool:
"""Valida os dígitos verificadores do CPF."""
if len(cpf) != 11 or not cpf.isdigit():
return False
if cpf == cpf[0] * 11:
return False
# Primeiro dígito
soma = sum(int(cpf[i]) * (10 - i) for i in range(9))
resto = (soma * 10) % 11
if resto == 10:
resto = 0
if resto != int(cpf[9]):
return False
# Segundo dígito
soma = sum(int(cpf[i]) * (11 - i) for i in range(10))
resto = (soma * 10) % 11
if resto == 10:
resto = 0
if resto != int(cpf[10]):
return False
return True
Aplicação principal (main.py)
from fastapi import FastAPI, HTTPException
from contextlib import asynccontextmanager
from models import ConsultaCPFRequest, ValidacaoCPFResponse
from services import cpfhub_service
from validators import validar_digitos_cpf
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
yield
# Shutdown
await cpfhub_service.fechar()
app = FastAPI(
title="Serviço de Validação de CPF",
description="Microserviço para validação de CPF usando CPFHub.io",
version="1.0.0",
lifespan=lifespan
)
@app.post("/validar-cpf", response_model=ValidacaoCPFResponse)
async def validar_cpf(request: ConsultaCPFRequest):
cpf = request.cpf
# Validação local dos dígitos verificadores
if not validar_digitos_cpf(cpf):
return ValidacaoCPFResponse(
cpf_valido=False,
mensagem="CPF com dígitos verificadores inválidos"
)
# Consulta à API do CPFHub.io
resultado = await cpfhub_service.consultar_cpf(cpf)
if not resultado.success:
return ValidacaoCPFResponse(
cpf_valido=False,
mensagem=resultado.erro or "CPF não encontrado"
)
dados = resultado.data
return ValidacaoCPFResponse(
cpf_valido=True,
nome=dados.name,
nascimento=dados.birthDate,
genero=dados.gender,
mensagem="CPF válido e verificado"
)
@app.get("/consultar-cpf/{cpf}")
async def consultar_cpf(cpf: str):
cpf_limpo = cpf.replace(".", "").replace("-", "")
if len(cpf_limpo) != 11:
raise HTTPException(
status_code=400,
detail="CPF deve ter 11 dígitos"
)
resultado = await cpfhub_service.consultar_cpf(cpf_limpo)
if not resultado.success:
raise HTTPException(
status_code=404,
detail=resultado.erro or "CPF não encontrado"
)
return resultado
@app.get("/health")
async def health_check():
return {"status": "ok", "service": "cpf-validation"}
Executando o serviço
uvicorn main:app --reload --port 8000
Após iniciar, acesse:
- Documentação interativa (Swagger): http://localhost:8000/docs
- Documentação alternativa (ReDoc): http://localhost:8000/redoc
- Health check: http://localhost:8000/health
Testando com cURL
curl -X POST http://localhost:8000/validar-cpf \
-H "Content-Type: application/json" \
-d '{"cpf": "123.456.789-00"}'
curl -X GET http://localhost:8000/consultar-cpf/12345678900
Adicionando cache com TTL
Para reduzir chamadas à API e melhorar a performance, implemente um cache simples:
from functools import lru_cache
from datetime import datetime, timedelta
# Cache simples em memória
_cache = {}
CACHE_TTL = timedelta(hours=1)
async def consultar_cpf_com_cache(cpf: str):
agora = datetime.now()
if cpf in _cache:
resultado, timestamp = _cache[cpf]
if agora - timestamp < CACHE_TTL:
return resultado
resultado = await cpfhub_service.consultar_cpf(cpf)
if resultado.success:
_cache[cpf] = (resultado, agora)
return resultado
Boas práticas
-
Variáveis de ambiente -- Nunca coloque a chave de API diretamente no código. Use .env e python-dotenv.
-
Validação em camadas -- Valide os dígitos localmente antes de chamar a API para economizar consultas.
-
Timeout configurável -- O timeout de 10 segundos é adequado para a API do CPFHub.io (resposta ~900ms).
-
Logging estruturado -- Registre todas as consultas para auditoria e diagnóstico.
-
Controle de consumo -- Se sua aplicação tem muitos usuários simultâneos, monitore o uso em app.cpfhub.io/settings/billing. O CPFHub.io não bloqueia ao atingir o limite — cobra R$0,15 por consulta adicional —, portanto o monitoramento proativo evita surpresas na fatura.
Perguntas frequentes
Como faço para que o microserviço FastAPI não exponha a chave de API ao cliente?
Use o módulo $env/static/private equivalente do Python: armazene a chave em variável de ambiente carregada com python-dotenv e nunca retorne a chave em nenhuma resposta da API. O cliente do frontend deve chamar apenas os endpoints do seu microserviço, nunca a API do CPFHub.io diretamente.
Qual a latência esperada da API do CPFHub.io em produção?
A latência típica é de aproximadamente 900ms por consulta. Configure o timeout do httpx em pelo menos 10 segundos para absorver variações de rede. O cache em memória com TTL de 1 hora elimina latência em consultas repetidas ao mesmo CPF dentro da mesma sessão.
Como o FastAPI lida com múltiplas requisições simultâneas à API de CPF?
FastAPI é assíncrono por padrão. Com httpx.AsyncClient configurado como singleton (instanciado uma vez no service), múltiplas requisições ao endpoint /validar-cpf são processadas em paralelo sem criar um novo cliente HTTP por requisição — o que é mais eficiente e reduz o consumo de recursos.
O que acontece se o limite de consultas mensais for atingido?
O CPFHub.io não bloqueia a API quando o limite do plano é atingido. Cada consulta adicional é cobrada a R$0,15 — a integração continua funcionando normalmente. Para controlar o consumo, acesse app.cpfhub.io/settings/billing e configure alertas de uso.
Conclusão
FastAPI e CPFHub.io formam uma combinação poderosa para construir microserviços de validação de CPF em Python. A tipagem forte do Pydantic, o suporte assíncrono do httpx e a documentação automática do FastAPI criam uma solução robusta, testável e bem documentada. O plano gratuito do CPFHub.io com 50 consultas por mês é perfeito para desenvolvimento, e os planos Pro (R$ 149/mês) e Corporativo atendem produção em qualquer escala.
Cadastre-se em cpfhub.io — 50 consultas mensais gratuitas, sem cartão de crédito — e suba seu microserviço de validação de CPF em Python ainda hoje.
CPFHub.io
Pronto para integrar a API?
50 consultas gratuitas para testar agora. Sem cartão de crédito. Acesso imediato à 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.



