Como Implementar Rate Limiting e Exponential Backoff ao Consumir APIs de CPF

# Python - Lendo headers de rate limit
import requests
import time

def consultar_com_awareness(cpf: str, api_key: str) -> dict:
    response = requests.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    headers={"x-api-key": api_key}
    )

    remaining = int(response.headers.get("X-RateLimit-Remaining", 100))
    reset_time = int(response.headers.get("X-RateLimit-Reset", 0))

    if remaining < 10:
    wait_time = max(0, reset_time - int(time.time()))
    print(f"Atenção: apenas {remaining} requisições restantes. "
    f"Reset em {wait_time}s")

    response.raise_for_status()
    return response.json()

import time
import threading
from collections import deque

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: float):
    self.max_requests = max_requests
    self.window = window_seconds
    self.timestamps = deque()
    self.lock = threading.Lock()

    def acquire(self):
    with self.lock:
    now = time.monotonic()

    # Remover timestamps fora da janela
    while self.timestamps and self.timestamps[0] <= now - self.window:
    self.timestamps.popleft()

    if len(self.timestamps) >= self.max_requests:
    wait_time = self.timestamps[0] + self.window - now
    time.sleep(wait_time)
    return self.acquire()

    self.timestamps.append(now)

class CPFClientComLimite:
    def __init__(self, api_key: str, max_rps: int = 10):
    self.api_key = api_key
    self.limiter = RateLimiter(max_requests=max_rps, window_seconds=1.0)
    self.session = requests.Session()
    self.session.headers["x-api-key"] = api_key

    def consultar(self, cpf: str) -> dict:
    self.limiter.acquire()
    response = self.session.get(
    f"https://api.cpfhub.io/cpf/{cpf}",
    timeout=15
    )
    response.raise_for_status()
    return response.json()

    def consultar_lote(self, cpfs: list) -> list:
    resultados = []
    for i, cpf in enumerate(cpfs):
    try:
    resultado = self.consultar(cpf)
    resultados.append({"cpf": cpf, "dados": resultado, "erro": None})
    except Exception as e:
    resultados.append({"cpf": cpf, "dados": None, "erro": str(e)})

    if (i + 1) % 50 == 0:
    print(f"Processados {i + 1}/{len(cpfs)} CPFs")

    return resultados

class RateLimiter {
    constructor(maxRequests, windowMs) {
    this.maxRequests = maxRequests;
    this.windowMs = windowMs;
    this.timestamps = [];
    this.queue = [];
    }

    async acquire() {
    return new Promise((resolve) => {
    this.queue.push(resolve);
    this.#process();
    });
    }

    #process() {
    const now = Date.now();
    this.timestamps = this.timestamps.filter(
    (t) => t > now - this.windowMs
    );

    while (this.queue.length > 0 && this.timestamps.length < this.maxRequests) {
    this.timestamps.push(now);
    const resolve = this.queue.shift();
    resolve();
    }

    if (this.queue.length > 0) {
    const oldestTimestamp = this.timestamps[0];
    const waitTime = oldestTimestamp + this.windowMs - now;
    setTimeout(() => this.#process(), Math.max(waitTime, 10));
    }
    }
}

class CPFClientComLimite {
    constructor(apiKey, maxRPS = 10) {
    this.apiKey = apiKey;
    this.limiter = new RateLimiter(maxRPS, 1000);
    this.baseUrl = "https://api.cpfhub.io/cpf";
    }

    async consultar(cpf) {
    await this.limiter.acquire();

    const response = await fetch(`${this.baseUrl}/${cpf.replace(/\D/g, "")}`, {
    headers: { "x-api-key": this.apiKey },
    });

    if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
    }

    return response.json();
    }

    async consultarLote(cpfs) {
    const resultados = [];
    for (const cpf of cpfs) {
    try {
    const dados = await this.consultar(cpf);
    resultados.push({ cpf, dados, erro: null });
    } catch (error) {
    resultados.push({ cpf, dados: null, erro: error.message });
    }
    }
    return resultados;
    }
}

import random
import time

def exponential_backoff_retry(
    func,
    max_tentativas: int = 5,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    jitter_strategy: str = "full"
):
    for tentativa in range(max_tentativas):
    try:
    return func()
    except Exception as e:
    if tentativa == max_tentativas - 1:
    raise

    # Calcular delay com backoff exponencial
    delay = min(base_delay * (2 ** tentativa), max_delay)

    # Aplicar jitter
    if jitter_strategy == "full":
    delay = random.uniform(0, delay)
    elif jitter_strategy == "equal":
    delay = delay / 2 + random.uniform(0, delay / 2)
    elif jitter_strategy == "decorrelated":
    delay = random.uniform(base_delay, delay * 3)

    print(f"Tentativa {tentativa + 1} falhou: {e}. "
    f"Aguardando {delay:.1f}s")
    time.sleep(delay)

# Uso com consulta de CPF
client = CPFClientComLimite(api_key="sua-chave", max_rps=10)

resultado = exponential_backoff_retry(
    lambda: client.consultar("12345678909"),
    max_tentativas=3,
    base_delay=1.0
)