Rate Limiter em Python Puro: O Guardião Que Impede Sua Automação de Se Tornar o Ataque DDoS Que Você Sempre Temeu

Semana passada, um script meu de web scraping — daqueles que roda “de boa” a cada 5 minutos — foi bloqueado permanentemente pela API que eu usava há dois anos. Motivo? Eu tinha esquecido de limitar a taxa de requisições numa nova thread que adicionei. O resultado: 847 requisições em 12 segundos. A API me respondeu com um 429 Too Many Requests e, na sequência, um 403 Forbidden eterno.

Perdi o acesso. Perdi os dados. Perdi a paciência.

Se você já escreveu qualquer automação que faz requisições HTTP — scraper, monitor, bot, integrador — você está a um while True mal pensado de se tornar o atacante da sua própria infraestrutura.

Hoje vou te mostrar como construir um Rate Limiter robusto em Python puro, sem pip install nada, com duas estratégias complementares: Token Bucket e Sliding Window Log. Código testado, pronto pra copiar, com decorador pra aplicar em qualquer função em 3 segundos.

Por que Rate Limiter local se já tem na API?

Ótima pergunta. Três razões:

1. A API pode não ter limiter configurado corretamente — e quando tem, o erro 429 vem acompanhado de delay arbitrário que quebra seu pipeline.

2. Você pode ter múltiplos scripts atacando o mesmo endpoint — o limiter da API vê “você”, não sabe que são 4 processos diferentes.

3. Proteger recursos locais também importa — banco SQLite, arquivos de log, threads de trabalho. Tudo tem limite.

A filosofia aqui é defesa em profundidade: você controla o fluxo na origem, não depende do servidor te bloquear.

Estratégia 1: Token Bucket — O Clássico Confiável

O Token Bucket funciona como uma caixa d’água com torneira: tokens entram em taxa constante, cada requisição consome um token, e se a caixa esvaziar, você espera.

import time
import threading


class TokenBucket:
    """
    Rate Limiter baseado em Token Bucket.
    
    - capacity: máximo de tokens acumuláveis
    - refill_rate: tokens adicionados por segundo
    - Exemplo: capacity=10, refill_rate=2 → até 10 req/s burst,
      sustentando 2 req/s continuamente
    """

    def __init__(self, capacity: float, refill_rate: float):
        if capacity <= 0 or refill_rate <= 0:
            raise ValueError("capacity e refill_rate devem ser positivos")
        
        self.capacity = capacity
        self.refill_rate = refill_rate
        self._tokens = capacity  # começa cheio
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()

    def _refill(self):
        """Recarrega tokens baseado no tempo elapsed."""
        now = time.monotonic()
        elapsed = now - self._last_refill
        tokens_to_add = elapsed * self.refill_rate
        self._tokens = min(self.capacity, self._tokens + tokens_to_add)
        self._last_refill = now

    def acquire(self, tokens: float = 1.0, block: bool = True) -> bool:
        """Consome tokens. Bloqueia se block=True e não houver tokens."""
        with self._lock:
            self._refill()
            
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            
            if not block:
                return False
            
            deficit = tokens - self._tokens
            wait_time = deficit / self.refill_rate
        
        time.sleep(wait_time)
        
        with self._lock:
            self._refill()
            self._tokens -= tokens
            return True

    @property
    def available_tokens(self) -> float:
        with self._lock:
            self._refill()
            return self._tokens

Testando na prática:

# Configuração: burst de 5 requisições, sustentando 2 por segundo
limiter = TokenBucket(capacity=5, refill_rate=2)

for i in range(8):
    start = time.monotonic()
    limiter.acquire()
    elapsed = time.monotonic() - start
    print(f"Req {i+1}: {'imediata' if elapsed < 0.01 else f'esperou {elapsed:.2f}s'}")

Saída:

Req 1: imediata
Req 2: imediata
Req 3: imediata
Req 4: imediata
Req 5: imediata
Req 6: esperou 0.50s
Req 7: esperou 0.50s
Req 8: esperou 0.50s

As 5 primeiras passaram de imediato (burst). A partir da sexta, o limiter segura o ritmo em 2/segundo — exatamente o refill_rate.

Estratégia 2: Sliding Window Log — Precisão Cirúrgica

O Token Bucket permite burst. Se você precisa de controle absoluto — “no máximo N requisições em qualquer janela de T segundos, sem exceção” — o Sliding Window Log é o caminho.

import time
import threading
from collections import deque


class SlidingWindowLog:
    """
    Rate Limiter baseado em Sliding Window Log.
    Garante no máximo max_requests em qualquer janela de window_seconds.
    """

    def __init__(self, max_requests: int, window_seconds: float):
        if max_requests <= 0 or window_seconds <= 0:
            raise ValueError("Parâmetros devem ser positivos")
        
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._log = deque()
        self._lock = threading.Lock()

    def _cleanup(self):
        cutoff = time.monotonic() - self.window_seconds
        while self._log and self._log[0] < cutoff:
            self._log.popleft()

    def acquire(self, block: bool = True) -> bool:
        while True:
            with self._lock:
                self._cleanup()
                
                if len(self._log) < self.max_requests:
                    self._log.append(time.monotonic())
                    return True
                
                if not block:
                    return False
                
                oldest = self._log[0]
                wait_time = (oldest + self.window_seconds) - time.monotonic()
            
            if wait_time > 0:
                time.sleep(wait_time)

    @property
    def current_count(self) -> int:
        with self._lock:
            self._cleanup()
            return len(self._log)

O Decorador Mágico: Proteja Qualquer Função em 3 Linhas

Aqui é onde a coisa fica elegante. Um decorador que aplica rate limiting transparente:

import functools


def rate_limit(limiter):
    """Decorador que aplica rate limiting a qualquer função."""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            limiter.acquire()
            return func(*args, **kwargs)
        return wrapper
    return decorator


# === Exemplo de uso real ===
bucket = TokenBucket(capacity=3, refill_rate=1)

@rate_limit(bucket)
def fetch_page(url: str) -> str:
    import urllib.request
    with urllib.request.urlopen(url, timeout=10) as resp:
        return resp.read().decode()


# Múltiplos limiters empilhados (global + por-endpoint)
global_bucket = TokenBucket(capacity=10, refill_rate=5)
endpoint_sliding = SlidingWindowLog(max_requests=30, window_seconds=60)

@rate_limit(global_bucket)
@rate_limit(endpoint_sliding)
def search_api(query: str) -> dict:
    url = f"https://api.example.com/search?q={query}"
    import urllib.request, json
    with urllib.request.urlopen(url, timeout=10) as resp:
        return json.loads(resp.read())

Sim, você pode empilhar decoradores. O limite mais restritivo sempre vence. É exatamente assim que APIs reais como GitHub e Stripe funcionam internamente.

Rate Limiter para Recursos Locais (não só HTTP)

O mesmo pattern protege qualquer recurso finito:

# Proteger writes em SQLite (evita SQLITE_BUSY)
db_write_limiter = TokenBucket(capacity=1, refill_rate=10)

@rate_limit(db_write_limiter)
def save_record(record: dict):
    with sqlite3.connect("data.db", timeout=5) as conn:
        conn.execute(
            "INSERT INTO records (data) VALUES (?)",
            (json.dumps(record),)
        )

# Proteger CPU em processamento de imagens
cpu_limiter = SlidingWindowLog(max_requests=5, window_seconds=1)

@rate_limit(cpu_limiter)
def process_image(path: str):
    # processamento pesado aqui
    pass

# Proteger disco (evita I/O saturado)
io_limiter = TokenBucket(capacity=50 * 1024 * 1024, refill_rate=10 * 1024 * 1024)
# 50MB burst, 10MB/s sustentável

def write_large_file(data: bytes, path: str):
    chunk_size = 1024 * 1024  # 1MB chunks
    for i in range(0, len(data), chunk_size):
        io_limiter.acquire(tokens=chunk_size)
        with open(path, "ab") as f:
            f.write(data[i:i + chunk_size])

Rate Limiter Compartilhado entre Processos (via arquivo)

Para múltiplos processos Python atacando o mesmo recurso:

import os, json, time, fcntl


class SharedRateLimiter:
    """Rate Limiter compartilhado entre processos via arquivo lock."""

    def __init__(self, lock_path: str, max_requests: int, window_seconds: float):
        self.lock_path = lock_path
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        
        if not os.path.exists(lock_path):
            with open(lock_path, "w") as f:
                json.dump([], f)

    def acquire(self):
        while True:
            with open(self.lock_path, "r+") as f:
                fcntl.flock(f.fileno(), fcntl.LOCK_EX)
                try:
                    timestamps = json.loads(f.read())
                    cutoff = time.monotonic() - self.window_seconds
                    timestamps = [t for t in timestamps if t >= cutoff]
                    
                    if len(timestamps) < self.max_requests:
                        timestamps.append(time.monotonic())
                        f.seek(0)
                        f.truncate()
                        json.dump(timestamps, f)
                        return
                    else:
                        oldest = min(timestamps)
                        wait = (oldest + self.window_seconds) - time.monotonic()
                finally:
                    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
            
            if wait > 0:
                time.sleep(wait)

🔥 O Perrengue do Olivetto: 847 Requisições Que Mataram Minha Conta

Era uma sexta-feira à noite. Eu tinha um scraper rodando há meses, coletando dados de uma API pública de preços. Funcionava lindo: cron job a cada 5 minutos, 3 requisições por execução.

Aí resolvi “otimizar” adicionando threads paralelas para buscar mais endpoints. Quatro threads, cada uma fazendo 3 requisições. Total esperado: 12 requisições a cada 5 minutos.

O que eu esqueci: as threads não tinham rate limiter individual. E o threading.Thread do Python não garante ordem de execução. Resultado: as 4 threads dispararam quase simultaneamente, cada uma fez suas 3 requisições, e o retry automático (que eu também tinha adicionado sem pensar) multiplicou tudo quando a API começou a responder devagar.

847 requisições em 12 segundos.

A API me bloqueou. Não foi um 429 temporário — foi ban permanente. Perdi 8 meses de dados históricos porque o endpoint de exportação também foi bloqueado.

Desde esse dia, toda função que toca rede passa pelo decorador @rate_limit. Sem exceção. Sem “ah, é só uma requisição”. O custo de adicionar o decorador é zero. O custo de esquecer pode ser meses de trabalho.

Qual Estratégia Escolher?

API com rate limit conhecido → Sliding Window — precisão absoluta, respeita a janela da API.

Automação geral / scraping → Token Bucket — burst controlado + throughput estável.

Múltiplos endpoints, um servidor → Token Bucket global + Sliding por endpoint — empilhamento cobre ambos os limites.

Recursos locais (DB, disco, CPU) → Token Bucket — simples, eficiente, permite burst quando faz sentido.

Multi-processo → SharedRateLimiter (arquivo) — coordenação via lock de arquivo.

E agora?

Qual foi o pior ban que você já levou por esquecer rate limiter? Ou pior: qual automação sua está rodando agora sem nenhum controle de taxa e você só vai descobrir quando quebrar?

Cola nos comentários que eu quero saber — e se tiver dúvida sobre qual estratégia usar no seu caso, manda o cenário que eu respondo.

Código testado em Python 3.11+ | Zero dependências externas | Pronto pra produção (com os devidos testes, claro)