Como Lidar com Erros HTTP ao Consumir APIs de CPF

import requests
import time
import logging

logger = logging.getLogger(__name__)

class CPFApiError(Exception):
    def __init__(self, message: str, status_code: int = None, retryable: bool = False):
    super().__init__(message)
    self.status_code = status_code
    self.retryable = retryable

class CPFClient:
    def __init__(self, api_key: str):
    self.session = requests.Session()
    self.session.headers["x-api-key"] = api_key
    self.base_url = "https://api.cpfhub.io/cpf"

    def consultar(self, cpf: str) -> dict:
    cpf_limpo = "".join(c for c in cpf if c.isdigit())

    try:
    response = self.session.get(
    f"{self.base_url}/{cpf_limpo}",
    timeout=15
    )
    except requests.exceptions.Timeout:
    raise CPFApiError("Timeout na requisição", retryable=True)
    except requests.exceptions.ConnectionError:
    raise CPFApiError("Erro de conexão com a API", retryable=True)

    return self._tratar_resposta(response)

    def _tratar_resposta(self, response: requests.Response) -> dict:
    handlers = {
    200: self._handle_success,
    400: self._handle_bad_request,
    401: self._handle_unauthorized,
    403: self._handle_forbidden,
    404: self._handle_not_found,
    }

    handler = handlers.get(response.status_code, self._handle_server_error)
    return handler(response)

    def _handle_success(self, response):
    data = response.json()
    if not data.get("success"):
    raise CPFApiError("CPF não encontrado na base de dados")
    return data

    def _handle_bad_request(self, response):
    raise CPFApiError("CPF inválido ou malformado", status_code=400)

    def _handle_unauthorized(self, response):
    raise CPFApiError("API key inválida ou ausente", status_code=401)

    def _handle_forbidden(self, response):
    raise CPFApiError("Sem permissão para este recurso", status_code=403)

    def _handle_not_found(self, response):
    raise CPFApiError("CPF não encontrado", status_code=404)

    def _handle_server_error(self, response):
    logger.error(f"Erro do servidor: {response.status_code}")
    raise CPFApiError(
    f"Erro interno da API ({response.status_code})",
    status_code=response.status_code,
    retryable=True
    )

class CPFApiError extends Error {
    constructor(message, statusCode = null, retryable = false) {
    super(message);
    this.name = "CPFApiError";
    this.statusCode = statusCode;
    this.retryable = retryable;
    }
}

class CPFClient {
    constructor(apiKey) {
    this.apiKey = apiKey;
    this.baseUrl = "https://api.cpfhub.io/cpf";
    }

    async consultar(cpf) {
    const cpfLimpo = cpf.replace(/\D/g, "");
    const controller = new AbortController();
    const timeoutId = setTimeout(() => controller.abort(), 15000);

    try {
    const response = await fetch(`${this.baseUrl}/${cpfLimpo}`, {
    headers: { "x-api-key": this.apiKey },
    signal: controller.signal,
    });

    clearTimeout(timeoutId);
    return this.#tratarResposta(response);
    } catch (error) {
    clearTimeout(timeoutId);
    if (error.name === "AbortError") {
    throw new CPFApiError("Timeout na requisição", null, true);
    }
    if (error instanceof CPFApiError) throw error;
    throw new CPFApiError(`Erro de conexão: ${error.message}`, null, true);
    }
    }

    async #tratarResposta(response) {
    const handlers = {
    400: () => new CPFApiError("CPF inválido ou malformado", 400),
    401: () => new CPFApiError("API key inválida ou ausente", 401),
    403: () => new CPFApiError("Sem permissão para este recurso", 403),
    404: () => new CPFApiError("CPF não encontrado", 404),
    };

    if (response.ok) {
    const data = await response.json();
    if (!data.success) {
    throw new CPFApiError("CPF não encontrado na base");
    }
    return data;
    }

    const handler = handlers[response.status];
    if (handler) throw handler();

    throw new CPFApiError(
    `Erro do servidor (${response.status})`,
    response.status,
    response.status >= 500
    );
    }
}

import time
import random

def consultar_com_retry(
    client: CPFClient,
    cpf: str,
    max_tentativas: int = 3,
    delay_base: float = 1.0
) -> dict:
    for tentativa in range(max_tentativas):
    try:
    return client.consultar(cpf)
    except CPFApiError as e:
    if not e.retryable or tentativa == max_tentativas - 1:
    raise

    delay = delay_base * (2 ** tentativa)
    jitter = random.uniform(0, delay * 0.1)
    tempo_espera = delay + jitter

    logger.warning(
    f"Tentativa {tentativa + 1}/{max_tentativas} falhou: {e}. "
    f"Retentando em {tempo_espera:.1f}s"
    )
    time.sleep(tempo_espera)

    raise CPFApiError("Todas as tentativas falharam")

async function consultarComRetry(client, cpf, maxTentativas = 3) {
    for (let tentativa = 0; tentativa < maxTentativas; tentativa++) {
    try {
    return await client.consultar(cpf);
    } catch (error) {
    if (!error.retryable || tentativa === maxTentativas - 1) {
    throw error;
    }

    const delay = Math.pow(2, tentativa) * 1000;
    const jitter = Math.random() * delay * 0.1;
    console.warn(
    `Tentativa ${tentativa + 1}/${maxTentativas} falhou. ` +
    `Retentando em ${((delay + jitter) / 1000).toFixed(1)}s`
    );
    await new Promise((r) => setTimeout(r, delay + jitter));
    }
    }
}