Monitorar consumo e custos da API de CPF com métricas

Monitorar o consumo da API de CPF é essencial para evitar surpresas de cobrança e garantir que a integração funcione dentro do esperado. A CPFHub.io não bloqueia quando a cota é esgotada -- cobra R$0,15 por consulta extra -- então sem visibilidade em tempo real, o volume pode escapar antes que qualquer alerta dispare. Este guia mostra como implementar métricas, alertas e dashboards para manter o controle completo da integração.

As três métricas essenciais

Para monitorar uma integração com API de CPF, concentre-se nestas três métricas:

1. Consumo de cota

Quantas consultas foram realizadas no período atual. Essencial para evitar surpresas de cobrança ou interrupção de serviço.

2. Latência de resposta

Tempo entre enviar a requisição e receber a resposta. A API da CPFHub.io tem latência média de aproximadamente 900ms -- valores consistentemente acima de 2 segundos indicam degradação que merece investigação.

3. Taxa de erro

Porcentagem de requisições que falharam (timeouts, erros 4xx/5xx). Um aumento repentino exige investigação imediata.

Implementando um wrapper com métricas

O primeiro passo é encapsular as chamadas à API em um wrapper que registra métricas automaticamente:

import time
import requests
import logging
from dataclasses import dataclass, field
from typing import Optional, Dict, List
from datetime import datetime, date

logger = logging.getLogger(__name__)

@dataclass
class MetricasAPI:
    """Armazena métricas de uso da API."""
    total_requisicoes: int = 0
    requisicoes_sucesso: int = 0
    requisicoes_erro: int = 0
    timeouts: int = 0
    latencias: List[float] = field(default_factory=list)
    consumo_diario: Dict[str, int] = field(default_factory=dict)

    @property
    def taxa_erro(self) -> float:
    if self.total_requisicoes == 0:
    return 0.0
    return (self.requisicoes_erro / self.total_requisicoes) * 100

    @property
    def latencia_media(self) -> float:
    if not self.latencias:
    return 0.0
    return sum(self.latencias) / len(self.latencias)

    @property
    def latencia_p95(self) -> float:
    if not self.latencias:
    return 0.0
    sorted_lat = sorted(self.latencias)
    idx = int(len(sorted_lat) * 0.95)
    return sorted_lat[idx]

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

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

class CPFClientMonitorado:
    """Cliente de API de CPF com monitoramento integrado."""

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

    def __init__(self, api_key: str, cota_mensal: int = 1000):
    self.api_key = api_key
    self.cota_mensal = cota_mensal
    self.metricas = MetricasAPI()
    self.session = requests.Session()
    self.session.headers.update({
    "x-api-key": self.api_key,
    "Accept": "application/json"
    })

    def consultar(self, cpf: str) -> Optional[Dict]:
    """Consulta CPF e registra métricas."""
    self.metricas.total_requisicoes += 1
    self.metricas.registrar_consumo_hoje()

    # Alerta de cota
    consumo = self.metricas.consumo_mes_atual()
    if consumo >= self.cota_mensal * 0.8:
    logger.warning(
    f"ALERTA: {consumo}/{self.cota_mensal} consultas usadas "
    f"({(consumo/self.cota_mensal)*100:.0f}%)"
    )

    inicio = time.time()

    try:
    response = self.session.get(
    f"{self.BASE_URL}/{cpf}",
    timeout=self.TIMEOUT
    )
    latencia = time.time() - inicio
    self.metricas.latencias.append(latencia)

    response.raise_for_status()
    dados = response.json()

    if dados.get("success"):
    self.metricas.requisicoes_sucesso += 1
    return dados
    else:
    self.metricas.requisicoes_erro += 1
    return None

    except requests.exceptions.Timeout:
    self.metricas.requisicoes_erro += 1
    self.metricas.timeouts += 1
    logger.error(f"Timeout ao consultar CPF")
    return None

    except requests.exceptions.RequestException as e:
    latencia = time.time() - inicio
    self.metricas.latencias.append(latencia)
    self.metricas.requisicoes_erro += 1
    logger.error(f"Erro na requisicao: {e}")
    return None

    def relatorio(self) -> str:
    """Gera relatório de métricas."""
    m = self.metricas
    return (
    f"=== Relatorio de Metricas ===\n"
    f"Total de requisicoes: {m.total_requisicoes}\n"
    f"Sucesso: {m.requisicoes_sucesso}\n"
    f"Erros: {m.requisicoes_erro}\n"
    f"Timeouts: {m.timeouts}\n"
    f"Taxa de erro: {m.taxa_erro:.1f}%\n"
    f"Latencia media: {m.latencia_media*1000:.0f}ms\n"
    f"Latencia P95: {m.latencia_p95*1000:.0f}ms\n"
    f"Consumo mensal: {m.consumo_mes_atual()}/{self.cota_mensal}\n"
    )

Configurando alertas

Alertas proativos evitam que problemas sejam descobertos tarde demais:

class AlertaConsumo:
    """Sistema de alertas baseado em métricas."""

    def __init__(self, cliente: CPFClientMonitorado):
    self.cliente = cliente

    def verificar_alertas(self) -> List[str]:
    """Retorna lista de alertas ativos."""
    alertas = []
    m = self.cliente.metricas

    # Alerta de cota
    consumo = m.consumo_mes_atual()
    percentual = (consumo / self.cliente.cota_mensal) * 100
    if percentual >= 90:
    alertas.append(f"CRITICO: Cota em {percentual:.0f}% ({consumo}/{self.cliente.cota_mensal})")
    elif percentual >= 80:
    alertas.append(f"AVISO: Cota em {percentual:.0f}% ({consumo}/{self.cliente.cota_mensal})")

    # Alerta de latência
    if m.latencia_media > 2.0:
    alertas.append(f"AVISO: Latencia media alta ({m.latencia_media*1000:.0f}ms)")

    # Alerta de taxa de erro
    if m.taxa_erro > 5.0:
    alertas.append(f"CRITICO: Taxa de erro em {m.taxa_erro:.1f}%")
    elif m.taxa_erro > 2.0:
    alertas.append(f"AVISO: Taxa de erro em {m.taxa_erro:.1f}%")

    return alertas

Exportando métricas para Prometheus

Se a sua equipe usa Prometheus e Grafana, exporte as métricas no formato padrão:

from prometheus_client import Counter, Histogram, Gauge

# Definir métricas
cpf_requests_total = Counter(
    "cpf_api_requests_total",
    "Total de requisicoes a API de CPF",
    ["status"]
)

cpf_request_duration = Histogram(
    "cpf_api_request_duration_seconds",
    "Latencia das requisicoes a API de CPF",
    buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)

cpf_quota_usage = Gauge(
    "cpf_api_quota_usage",
    "Consumo atual da cota mensal"
)

Monitorando via cURL

Para verificações rápidas sem código adicional, use cURL para medir latência:

curl -o /dev/null -s -w "Status: %{http_code}\nDNS: %{time_namelookup}s\nConexao: %{time_connect}s\nTLS: %{time_appconnect}s\nTransferencia: %{time_starttransfer}s\nTotal: %{time_total}s\n" \
    "https://api.cpfhub.io/cpf/12345678900" \
    -H "x-api-key: SUA_CHAVE_API" \
    -H "Accept: application/json" \
    --connect-timeout 10 \
    --max-time 30

Saída esperada:

Status: 200
DNS: 0.023s
Conexao: 0.045s
TLS: 0.112s
Transferencia: 0.934s
Total: 0.935s

Dashboards recomendados

Organize as métricas em três dashboards:

Dashboard operacional

Requisições por minuto (taxa de throughput).
Latência média, P50, P95 e P99.
Taxa de erro por tipo (timeout, 4xx, 5xx).

Dashboard de consumo

Consultas realizadas vs. cota disponível.
Projeção de consumo para o restante do mês.
Consumo por dia da semana (identifica padrões).

Dashboard de custos

Custo por consulta realizada.
Custo acumulado no mês.
Comparativo com meses anteriores.

Logs estruturados

Além de métricas, mantenha logs estruturados para debugging:

import json

def log_requisicao(cpf_masked, status_code, latencia, sucesso):
    """Registra log estruturado de uma requisição."""
    log_entry = {
    "timestamp": datetime.utcnow().isoformat(),
    "service": "cpf-api",
    "cpf_masked": cpf_masked,
    "status_code": status_code,
    "latency_ms": round(latencia * 1000),
    "success": sucesso
    }
    logger.info(json.dumps(log_entry))

# Exemplo de saída
# {"timestamp": "2026-08-21T14:30:00", "service": "cpf-api",
# "cpf_masked": "123***00", "status_code": 200,
# "latency_ms": 892, "success": true}

Note que o CPF é mascarado no log para conformidade com a LGPD.

Automatizando relatórios semanais

def gerar_relatorio_semanal(cliente: CPFClientMonitorado) -> str:
    """Gera relatório semanal de consumo."""
    m = cliente.metricas
    consumo = m.consumo_mes_atual()
    percentual = (consumo / cliente.cota_mensal) * 100

    return (
    f"Relatorio Semanal - API de CPF\n"
    f"Periodo: ultima semana\n"
    f"Consultas realizadas: {m.total_requisicoes}\n"
    f"Taxa de sucesso: {100 - m.taxa_erro:.1f}%\n"
    f"Latencia media: {m.latencia_media*1000:.0f}ms\n"
    f"Consumo mensal: {consumo}/{cliente.cota_mensal} ({percentual:.0f}%)\n"
    f"Projecao mensal: ~{int(consumo / max(date.today().day, 1) * 30)} consultas\n"
    )

Perguntas frequentes

Qual é a latência média da API de CPF e como monitorá-la?

A latência média da API da CPFHub.io é de aproximadamente 900ms. Monitore o P95 (percentil 95) além da média -- uma média saudável pode esconder picos que afetam a experiência do usuário. Configure um alerta quando o P95 ultrapassar 2 segundos consecutivos.

O que acontece quando a cota mensal da API é esgotada?

A CPFHub.io não bloqueia as requisições quando a cota é atingida. Cada consulta além das incluídas no plano é cobrada a R$0,15 automaticamente. O plano Gratuito inclui 50 consultas/mês e o plano Pro inclui 1.000 consultas por R$149/mês. Por isso, monitorar o consumo com alertas em 80% e 90% da cota é fundamental.

Como garantir conformidade com a LGPD ao registrar logs de consulta?

Nunca registre o CPF completo em logs. Use mascaramento -- por exemplo, 123***00 -- e aplique controle de acesso ao sistema de logs. A ANPD orienta que dados de identificação devem ser tratados com o princípio da necessidade, e logs com CPF completo raramente são necessários para diagnóstico.

Quais ferramentas são recomendadas para dashboards de observabilidade da API de CPF?

Prometheus + Grafana é a combinação mais comum para exportar e visualizar métricas de integração. O Grafana oferece painéis prontos para latência (histogramas), taxa de erro (alertas por threshold) e consumo acumulado (séries temporais). Para equipes menores, um relatório semanal gerado automaticamente em Python já resolve a maioria dos casos.

Conclusão

Monitorar o consumo e a saúde da integração com API de CPF não é um luxo -- é uma necessidade operacional. Sem observabilidade, problemas de latência, erros silenciosos e estouro de cota passam despercebidos até causarem impacto direto no negócio.

Com o wrapper monitorado, os alertas e os dashboards apresentados

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 monitorar consumo e custos com observabilidade