Como implementar health check para monitorar disponibilidade da API de CPF

curl -X GET https://api.cpfhub.io/cpf/00000000000 \
    -H "x-api-key: SUA_CHAVE_DE_API" \
    -H "Accept: application/json" \
    --max-time 15 \
    -w "\nHTTP_CODE: %{http_code}\nTIME: %{time_total}s\n" \
    -o /dev/null -s

import requests
import time
from datetime import datetime
from enum import Enum

class HealthStatus(Enum):
    HEALTHY = 'healthy'
    DEGRADED = 'degraded'
    UNHEALTHY = 'unhealthy'

class CPFHealthChecker:
    def __init__(self, api_key, latencia_limite_ms=3000):
    self.api_key = api_key
    self.latencia_limite = latencia_limite_ms
    self.session = requests.Session()
    self.session.headers.update({
    'x-api-key': api_key,
    'Accept': 'application/json'
    })

    def verificar(self):
    resultado = {
    'timestamp': datetime.utcnow().isoformat(),
    'status': HealthStatus.UNHEALTHY.value,
    'latencia_ms': None,
    'http_status': None,
    'resposta_valida': False,
    'detalhes': ''
    }

    try:
    inicio = time.perf_counter()
    response = self.session.get(
    'https://api.cpfhub.io/cpf/00000000000',
    timeout=15
    )
    fim = time.perf_counter()

    latencia = (fim - inicio) * 1000
    resultado['latencia_ms'] = round(latencia, 2)
    resultado['http_status'] = response.status_code

    # Verificar se a resposta e JSON valido
    try:
    data = response.json()
    resultado['resposta_valida'] = 'success' in data
    except ValueError:
    resultado['detalhes'] = 'Resposta nao e JSON valido'
    return resultado

    # Determinar status
    if response.status_code in [200, 404] and resultado['resposta_valida']:
    if latencia <= self.latencia_limite:
    resultado['status'] = HealthStatus.HEALTHY.value
    else:
    resultado['status'] = HealthStatus.DEGRADED.value
    resultado['detalhes'] = f'Latencia acima do limite ({latencia:.0f}ms > {self.latencia_limite}ms)'
    elif response.status_code == 401:
    resultado['detalhes'] = 'API key invalida'
    else:
    resultado['detalhes'] = f'Status inesperado: {response.status_code}'

    except requests.exceptions.Timeout:
    resultado['detalhes'] = 'Timeout na requisicao'
    except requests.exceptions.ConnectionError:
    resultado['detalhes'] = 'Erro de conexao'
    except Exception as e:
    resultado['detalhes'] = str(e)

    return resultado

# Uso
checker = CPFHealthChecker('SUA_CHAVE_DE_API')
status = checker.verificar()
print(f'Status: {status["status"]}')
print(f'Latencia: {status["latencia_ms"]}ms')

class CPFHealthChecker {
    constructor(apiKey, latenciaLimiteMs = 3000) {
    this.apiKey = apiKey;
    this.latenciaLimite = latenciaLimiteMs;
    }

    async verificar() {
    const resultado = {
    timestamp: new Date().toISOString(),
    status: 'unhealthy',
    latenciaMs: null,
    httpStatus: null,
    respostaValida: false,
    detalhes: ''
    };

    try {
    const inicio = performance.now();

    const response = await fetch('https://api.cpfhub.io/cpf/00000000000', {
    method: 'GET',
    headers: {
    'x-api-key': this.apiKey,
    'Accept': 'application/json'
    },
    signal: AbortSignal.timeout(15000)
    });

    const fim = performance.now();
    const latencia = fim - inicio;

    resultado.latenciaMs = Math.round(latencia * 100) / 100;
    resultado.httpStatus = response.status;

    const data = await response.json();
    resultado.respostaValida = 'success' in data;

    if ([200, 404].includes(response.status) && resultado.respostaValida) {
    resultado.status = latencia <= this.latenciaLimite ? 'healthy' : 'degraded';
    }
    } catch (error) {
    resultado.detalhes = error.message;
    }

    return resultado;
    }
}

// Uso
const checker = new CPFHealthChecker(process.env.CPFHUB_API_KEY);
const status = await checker.verificar();
console.log(JSON.stringify(status, null, 2));

# Verificar a cada 5 minutos
*/5 * * * * /usr/bin/python3 /opt/scripts/health_check_cpf.py >> /var/log/cpf_health.log 2>&1

const INTERVALO_MS = 5 * 60 * 1000; // 5 minutos

const checker = new CPFHealthChecker(process.env.CPFHUB_API_KEY);

setInterval(async () => {
    const status = await checker.verificar();
    console.log(`[${status.timestamp}] Status: ${status.status} | Latencia: ${status.latenciaMs}ms`);

    if (status.status === 'unhealthy') {
    // Disparar alerta (e-mail, Slack, PagerDuty)
    await enviarAlerta(status);
    }
}, INTERVALO_MS);

from flask import Flask, jsonify

app = Flask(__name__)
checker = CPFHealthChecker('SUA_CHAVE_DE_API')

@app.route('/health')
def health():
    status_cpfhub = checker.verificar()

    saude_geral = 'healthy'
    if status_cpfhub['status'] == 'unhealthy':
    saude_geral = 'unhealthy'
    elif status_cpfhub['status'] == 'degraded':
    saude_geral = 'degraded'

    return jsonify({
    'status': saude_geral,
    'dependencias': {
    'cpfhub_api': status_cpfhub
    }
    }), 200 if saude_geral == 'healthy' else 503