Como consumir API de CPF em Google Cloud Functions com Python

Aprenda a criar uma Cloud Function em Python que consome a API de CPF da CPFHub.io com tratamento de erros, cache e deploy automatizado.

Redação CPFHub.io
Redação CPFHub.io
··7 min de leitura
Como consumir API de CPF em Google Cloud Functions com Python

Google Cloud Functions é a plataforma serverless do Google que executa código Python em resposta a eventos HTTP sem gerenciar servidores. Integrar a API da CPFHub.io nesse ambiente permite criar endpoints de validação de CPF que escalam automaticamente, com resposta em torno de ~300ms e custo proporcional ao uso real. A autenticação usa o header x-api-key e o endpoint segue o padrão GET https://api.cpfhub.io/cpf/{CPF_NUMBER}.


Estrutura do projeto

Uma Cloud Function em Python requer no mínimo dois arquivos:

cpf-validator/
    main.py # Funcao principal
    requirements.txt # Dependencias

Para projetos mais complexos, podemos adicionar módulos auxiliares:

cpf-validator/
    main.py
    cpfhub_service.py
    validators.py
    requirements.txt

Serviço de consulta de CPF

Primeiro, criamos o módulo que encapsula a comunicação com a API da CPFHub.io

# cpfhub_service.py
import requests
import os
import re
from google.cloud import secretmanager

TIMEOUT_SECONDS = 10
API_URL = "https://api.cpfhub.io/cpf"

_cached_api_key = None

def get_api_key():
    """Recupera a chave de API do Secret Manager com cache."""
    global _cached_api_key

    if _cached_api_key:
    return _cached_api_key

    # Tentar variavel de ambiente primeiro (desenvolvimento local)
    api_key = os.environ.get("CPFHUB_API_KEY")

    if not api_key:
    # Buscar no Secret Manager (producao)
    client = secretmanager.SecretManagerServiceClient()
    project_id = os.environ.get("GCP_PROJECT")
    secret_name = f"projects/{project_id}/secrets/cpfhub-api-key/versions/latest"

    response = client.access_secret_version(name=secret_name)
    api_key = response.payload.data.decode("utf-8")

    _cached_api_key = api_key
    return api_key

def limpar_cpf(cpf_raw):
    """Remove caracteres nao numericos do CPF."""
    return re.sub(r"\D", "", cpf_raw)

def validar_formato_cpf(cpf):
    """Valida formato basico do CPF."""
    if len(cpf) != 11:
    return False, "CPF deve conter 11 digitos"

    if len(set(cpf)) == 1:
    return False, "CPF invalido (todos os digitos iguais)"

    return True, None

def consultar_cpf(cpf_numero):
    """Consulta CPF na API CPFHub.io."""
    cpf_limpo = limpar_cpf(cpf_numero)

    valido, erro = validar_formato_cpf(cpf_limpo)
    if not valido:
    return {"sucesso": False, "erro": erro}

    api_key = get_api_key()

    headers = {
    "x-api-key": api_key,
    "Accept": "application/json"
    }

    try:
    response = requests.get(
    f"{API_URL}/{cpf_limpo}",
    headers=headers,
    timeout=TIMEOUT_SECONDS
    )
    response.raise_for_status()

    dados = response.json()

    if not dados.get("success"):
    return {"sucesso": False, "erro": "CPF nao encontrado"}

    return {
    "sucesso": True,
    "dados": {
    "cpf": dados["data"]["cpf"],
    "nome": dados["data"]["name"],
    "genero": dados["data"]["gender"],
    "data_nascimento": dados["data"]["birthDate"],
    "dia": dados["data"]["day"],
    "mes": dados["data"]["month"],
    "ano": dados["data"]["year"]
    }
    }

    except requests.Timeout:
    return {"sucesso": False, "erro": "Timeout na consulta de CPF"}
    except requests.ConnectionError:
    return {"sucesso": False, "erro": "Erro de conexao com a API"}
    except requests.HTTPError as e:
    return {"sucesso": False, "erro": f"Erro HTTP: {e.response.status_code}"}
    except Exception as e:
    return {"sucesso": False, "erro": f"Erro inesperado: {str(e)}"}

Função principal (HTTP trigger)

A Cloud Function HTTP recebe um objeto Request do Flask e retorna uma resposta:

# main.py
import functions_framework
from flask import jsonify, make_response
from cpfhub_service import consultar_cpf, limpar_cpf

CORS_HEADERS = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type, Authorization",
    "Access-Control-Max-Age": "3600"
}

def criar_resposta(dados, status_code=200):
    """Cria resposta HTTP com headers CORS."""
    response = make_response(jsonify(dados), status_code)
    for key, value in CORS_HEADERS.items():
    response.headers[key] = value
    return response

@functions_framework.http
def validar_cpf(request):
    """Cloud Function para validacao de CPF.

    Args:
    request: Objeto Flask Request.

    Returns:
    Resposta JSON com dados do CPF ou mensagem de erro.
    """
    # Tratar preflight CORS
    if request.method == "OPTIONS":
    return criar_resposta({})

    if request.method != "GET":
    return criar_resposta(
    {"erro": "Metodo nao permitido. Use GET."},
    405
    )

    # Extrair CPF da URL ou query string
    cpf = None

    # Tentar extrair do path: /cpf/12345678901
    path = request.path.strip("/")
    partes = path.split("/")
    if len(partes) >= 2 and partes[0] == "cpf":
    cpf = partes[1]

    # Fallback: query string ?cpf=12345678901
    if not cpf:
    cpf = request.args.get("cpf")

    if not cpf:
    return criar_resposta(
    {"erro": "CPF e obrigatorio. Use /cpf/{numero} ou ?cpf={numero}"},
    400
    )

    cpf_limpo = limpar_cpf(cpf)

    if len(cpf_limpo) != 11:
    return criar_resposta(
    {"erro": "CPF deve conter 11 digitos"},
    400
    )

    resultado = consultar_cpf(cpf_limpo)

    if not resultado["sucesso"]:
    status = 404 if "nao encontrado" in resultado.get("erro", "") else 500
    return criar_resposta({"erro": resultado["erro"]}, status)

    return criar_resposta({
    "valido": True,
    "dados": resultado["dados"]
    })

Dependências e configuração

O arquivo requirements.txt:

functions-framework==3.*
requests>=2.31.0,<3.0.0
google-cloud-secret-manager>=2.16.0,<3.0.0

Deploy da Cloud Function

Deploy usando o Google Cloud CLI. Consulte a documentação oficial do Google Cloud Functions para detalhes sobre flags e configurações avançadas:

gcloud functions deploy validar-cpf \
    --gen2 \
    --runtime=python312 \
    --region=southamerica-east1 \
    --source=. \
    --entry-point=validar_cpf \
    --trigger-http \
    --allow-unauthenticated \
    --memory=256MB \
    --timeout=30s \
    --set-env-vars="GCP_PROJECT=meu-projeto"

A região southamerica-east1 (São Paulo) minimiza a latência para usuários brasileiros. Combinada com o tempo de resposta de ~300ms da API da CPFHub.io, a resposta total fica bem abaixo de 1 segundo na maioria dos cenários.


Gerenciamento de segredos

Para armazenar a chave de API de forma segura no Google Cloud:

# Criar o segredo
echo -n "sua_chave_cpfhub" | gcloud secrets create cpfhub-api-key \
    --data-file=- \
    --replication-policy=user-managed \
    --locations=southamerica-east1

# Conceder acesso a Cloud Function
gcloud secrets add-iam-policy-binding cpfhub-api-key \
    --member="serviceAccount:meu-projeto@appspot.gserviceaccount.com" \
    --role="roles/secretmanager.secretAccessor"

Testes locais

O Functions Framework permite testar localmente antes do deploy:

# Instalar dependencias
pip install -r requirements.txt

# Definir variavel de ambiente para teste local
export CPFHUB_API_KEY="sua_chave_de_teste"

# Iniciar servidor local
functions-framework --target=validar_cpf --port=8080

Teste com curl:

curl "http://localhost:8080/cpf/12345678901"

Cache com Memorystore ou Firestore

Para reduzir chamadas à API da CPFHub.io e melhorar a performance, implemente cache usando Firestore:

from google.cloud import firestore
from datetime import datetime, timedelta

db = firestore.Client()
CACHE_TTL_HOURS = 24

def buscar_cache(cpf):
    """Busca resultado em cache no Firestore."""
    doc_ref = db.collection("cpf_cache").document(cpf)
    doc = doc_ref.get()

    if doc.exists:
    dados = doc.to_dict()
    expiracao = dados.get("expira_em")

    if expiracao and expiracao > datetime.utcnow():
    return dados.get("resultado")

    return None

def salvar_cache(cpf, resultado):
    """Salva resultado no cache do Firestore."""
    doc_ref = db.collection("cpf_cache").document(cpf)
    doc_ref.set({
    "resultado": resultado,
    "expira_em": datetime.utcnow() + timedelta(hours=CACHE_TTL_HOURS),
    "atualizado_em": datetime.utcnow()
    })

Monitoramento e alertas

Configure alertas no Cloud Monitoring para acompanhar a saúde da integração:

  • Taxa de erro — alerte quando a taxa de erros 5xx ultrapassar 5% em 5 minutos.
  • Latência — alerte quando o p99 de latência ultrapassar 2 segundos.
  • Invocações — monitore o volume para prever quando migrar para o plano Pro da CPFHub.io.
  • Cold starts — acompanhe a frequência para avaliar se Provisioned Instances são necessárias.

Perguntas frequentes

Como funciona o rate limit da API CPFHub.io no plano gratuito?

No plano gratuito, o rate limit é de 1 requisição a cada 2 segundos. Ultrapassar essa taxa retorna HTTP 429 — mas isso é diferente de esgotar a cota mensal. Se você consumir as 50 consultas do plano gratuito, a API não bloqueia: cada consulta extra é cobrada a R$0,15. Para evitar surpresas, monitore o consumo no painel e considere implementar cache local.

Cloud Functions é uma boa opção para validação de CPF em alta frequência?

Sim, especialmente quando combinada com cache no Firestore. O modelo serverless escala automaticamente com a demanda, e o cache elimina chamadas repetidas para o mesmo CPF. Para volumes muito altos, o plano Pro da CPFHub.io (R$149/mês, 1.000 consultas incluídas) oferece rate limit de 1 req/s — o dobro do plano gratuito.

É seguro armazenar a chave de API em variáveis de ambiente no Cloud Functions?

Variáveis de ambiente no Cloud Functions são uma solução aceitável para desenvolvimento, mas para produção o recomendado é o Google Secret Manager. Ele criptografa o segredo em repouso, controla o acesso via IAM e registra cada acesso em audit logs — atendendo às boas práticas de segurança e aos requisitos de conformidade com a LGPD.

O que acontece se a API da CPFHub.io ficar indisponível durante a execução da função?

O código usa requests.Timeout e requests.ConnectionError para capturar essas situações e retornar mensagens de erro claras. Uma estratégia complementar é implementar retry com backoff exponencial usando a biblioteca tenacity, garantindo resiliência sem sobrecarregar a API em momentos de instabilidade.


Conclusão

Google Cloud Functions com Python oferecem uma base sólida para validação de CPF em ambiente serverless. A integração com Secret Manager resolve a questão de segurança das credenciais, o Functions Framework simplifica os testes locais, e a escolha da região São Paulo reduz a latência para usuários brasileiros. Com cache no Firestore e monitoramento ativo, você tem uma solução resiliente que escala sem esforço de operação.

Se você ainda não testou a API da CPFHub.io, o plano gratuito oferece 50 consultas por mês sem cartão de crédito — tempo suficiente para validar a integração completa. Crie sua conta em cpfhub.io e comece a consultar em menos de 30 minutos.

CPFHub.io

Pronto para integrar a API?

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

Redação CPFHub.io

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.

WhatsAppFale conosco via WhatsApp