Token Bucket Rate Limiter em Python Puro: O Algoritmo Que Limita Requisições na Sua API e Impede Que Um Único Usuário Derrube o Servidor (Sem Redis, Sem Nginx)
Você já teve uma API derrubada porque um único cliente mandou 10.000 requisições em 3 segundos? Ou pior — já foi esse cliente sem saber, porque seu script de automação não tinha limite de velocidade?
Rate limiting não é luxo. É sobrevivência. E você não precisa de Redis, Nginx, ou um serviço externo pra implementar um. O algoritmo Token Bucket cabe em 40 linhas de Python puro — e funciona tão bem quanto as soluções de produção que custam centenas de dólares por mês.
O Que É Token Bucket (E Por Que Ele É Melhor Que Um Simples Contador)
A maioria dos rate limiters ingênuos funciona assim: conta requisições numa janela de tempo e bloqueia quando passa do limite. Simples. E estúpido.
O problema? Um contador fixo cria uma parede abrupta. Você pode fazer 99 requisições no segundo 1 e zero no segundo 2. Mas se tentar 100 no segundo 1, é bloqueado — mesmo que o servidor esteja ocioso.
O Token Bucket é mais inteligente. Ele funciona como um balde com furo:
- Tokens são adicionados ao balde numa taxa constante (ex: 10 tokens por segundo)
- O balde tem capacidade máxima (ex: 20 tokens)
- Cada requisição consome 1 token
- Se o balde está vazio, a requisição é negada
- Se o balde está cheio, tokens novos são descartados
Isso permite bursts controlados. Um cliente pode mandar 20 requisições de uma vez (se o balde estava cheio), mas depois precisa esperar os tokens se regenerarem. É justo, é previsível, e é o mesmo algoritmo que o AWS API Gateway usa internamente.
A Implementação: 40 Linhas Que Salvam Seu Servidor
import time
import threading
class TokenBucket:
"""
Rate limiter baseado em Token Bucket.
Args:
rate: tokens adicionados por segundo
capacity: máximo de tokens no balde
"""
def __init__(self, rate: float, capacity: float):
self.rate = rate
self.capacity = capacity
self.tokens = capacity # Começa cheio
self.last_refill = time.monotonic()
self._lock = threading.Lock()
def _refill(self):
"""Adiciona tokens baseado no tempo elapsed desde o último refill."""
now = time.monotonic()
elapsed = now - self.last_refill
new_tokens = elapsed * self.rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
def consume(self, tokens: int = 1) -> bool:
"""
Tenta consumir tokens. Retorna True se conseguiu, False se negado.
Thread-safe via lock.
"""
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def wait_and_consume(self, tokens: int = 1, timeout: float = None) -> bool:
"""
Espera até ter tokens disponíveis e consome.
Retorna False se timeout foi atingido.
"""
deadline = None if timeout is None else time.monotonic() + timeout
while True:
with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
# Calcula quanto tempo até ter tokens suficientes
deficit = tokens - self.tokens
wait_time = deficit / self.rate
# Verifica timeout
if deadline is not None:
remaining = deadline - time.monotonic()
if remaining <= 0:
return False
wait_time = min(wait_time, remaining)
time.sleep(wait_time)
@property
def available_tokens(self) -> float:
"""Quantos tokens estão disponíveis agora."""
with self._lock:
self._refill()
return self.tokens
@property
def retry_after(self) -> float:
"""Segundos até o próximo token disponível."""
with self._lock:
self._refill()
if self.tokens >= 1:
return 0.0
return (1 - self.tokens) / self.rate
Como Usar: Do Script Simples ao Servidor HTTP
Exemplo 1: Limitando um Scraper
# Scraper que respeita rate limit do site
# 2 requisições por segundo, burst de 5
limiter = TokenBucket(rate=2.0, capacity=5)
urls = ["https://api.example.com/data/" + str(i) for i in range(100)]
for url in urls:
if limiter.wait_and_consume(timeout=10):
# Faz a requisição
print(f"Buscando {url}")
# response = urllib.request.urlopen(url)
else:
print(f"Timeout esperando por {url}")
break
Exemplo 2: Middleware HTTP com Rate Limiting por IP
from http.server import HTTPServer, BaseHTTPRequestHandler
from collections import defaultdict
# Um balde por IP — 10 req/s, burst de 20
buckets = defaultdict(lambda: TokenBucket(rate=10, capacity=20))
class RateLimitedHandler(BaseHTTPRequestHandler):
def do_GET(self):
client_ip = self.client_address[0]
bucket = buckets[client_ip]
if not bucket.consume():
# 429 Too Many Requests
self.send_response(429)
self.send_header('Retry-After', str(int(bucket.retry_after) + 1))
self.send_header('Content-Type', 'application/json')
self.end_headers()
error = json.dumps({
"error": "Rate limit exceeded",
"retry_after": bucket.retry_after,
"available_tokens": bucket.available_tokens
})
self.wfile.write(error.encode())
return
# Requisição permitida
self.send_response(200)
self.send_header('X-RateLimit-Remaining', str(int(bucket.available_tokens)))
self.send_header('Content-Type', 'application/json')
self.end_headers()
response = json.dumps({"message": "OK", "ip": client_ip})
self.wfile.write(response.encode())
def log_message(self, format, *args):
pass # Silencia logs padrão
# server = HTTPServer(('0.0.0.0', 8080), RateLimitedHandler)
# server.serve_forever()
Exemplo 3: Decorator para Qualquer Função
import functools
def rate_limited(rate: float, capacity: float):
"""Decorator que aplica rate limiting a qualquer função."""
bucket = TokenBucket(rate=rate, capacity=capacity)
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
bucket.wait_and_consume()
return func(*args, **kwargs)
wrapper._bucket = bucket # Expõe o bucket para inspeção
return wrapper
return decorator
@rate_limited(rate=5, capacity=10)
def send_email(to: str, subject: str, body: str):
"""Envia email respeitando limite de 5/s."""
print(f"Email enviado para {to}: {subject}")
# lógica real de envio aqui
# Uso:
for i in range(50):
send_email(f"user{i}@example.com", "Newsletter", "Conteúdo...")
# Automaticamente espera se passar do limite
O Perrengue do Olivetto: Quando Esqueci o Lock e Quebrei Tudo
A primeira versão desse Token Bucket que escrevi não tinha threading.Lock. Parecia funcionar perfeitamente nos testes — single-thread, sequencial, tudo verde.
Aí coloquei em produção com um servidor HTTP multi-thread. Duas threads tentaram consumir o último token ao mesmo tempo. Ambas leram self.tokens = 0.5, ambas acharam que tinham o suficiente (0.5 >= 0.5), ambas decrementaram. Resultado: self.tokens = -0.5.
Tokens negativos. O balde furou. O rate limiter deixou passar o dobro de requisições que deveria, e o servidor upstream — que era uma API com plano gratuito de 1.000 req/dia — me cobrou o excesso.
A lição? Qualquer estado compartilhado entre threads precisa de lock. Sem exceção. Sem “ah, mas é só uma leitura rápida”. A race condition não liga pro seu otimismo.
Token Bucket vs. Leaky Bucket vs. Fixed Window: Qual Usar?
Cada algoritmo tem seu lugar:
Fixed Window (contador por janela de tempo) — mais simples de implementar, mas tem o problema da borda: um cliente pode mandar o dobro do limite na transição entre janelas. Bom pra quando precisão não é crítica.
Leaky Bucket (balde furado) — taxa de saída constante, independente da entrada. Ideal pra proteger backends que não aguentam variação de carga. Mas não permite bursts, o que pode frustrar usuários legítimos.
Token Bucket — o melhor dos dois mundos. Permite bursts até o limite do balde, mas mantém a taxa média sob controle. É o que a maioria dos serviços de produção usa (AWS, Stripe, GitHub API).
Limitações e Quando Você Precisa de Algo Mais
Essa implementação é in-memory. Funciona perfeitamente para:
- Servidores single-instance
- Scripts e automações pessoais
- Proteção de APIs internas
- Desenvolvimento e testes
Mas se você tem múltiplas instâncias do servidor (load balancer na frente de 3 máquinas), cada instância tem seu próprio balde. O rate limit efetivo é multiplicado pelo número de instâncias.
Para cenários distribuídos, você precisa de um store compartilhado. Opções:
- Redis com módulo RedisBloom ou scripts Lua atomics
- Memcached com increment atomic
- DynamoDB com conditional writes
A lógica do Token Bucket é a mesma. Só muda onde os tokens são armazenados.
Headers HTTP Que Todo Rate Limiter Deve Retornar
Quando um cliente é limitado, ele merece saber quanto esperar. Os headers padrão da indústria são:
# Headers informativos (em toda resposta)
X-RateLimit-Limit: 20 # Capacidade do balde
X-RateLimit-Remaining: 15 # Tokens disponíveis
X-RateLimit-Reset: 1.5 # Segundos até balde cheio
# Header de bloqueio (apenas em 429)
Retry-After: 2 # Segundos até tentar de novo
O GitHub, Stripe e Twitter usam exatamente esses headers. Seus clientes (humanos ou scripts) já sabem como interpretá-los.
Bônus: Rate Limiter com Limpeza Automática de Buckets Antigos
Se você cria um bucket por IP, precisa limpar os que não são usados há muito tempo. Senão, a memória cresce infinitamente:
class BucketRegistry:
"""Gerencia múltiplos Token Buckets com limpeza automática."""
def __init__(self, rate: float, capacity: float, ttl: float = 3600):
self.rate = rate
self.capacity = capacity
self.ttl = ttl # Segundos sem uso antes de limpar
self._buckets = {} # key -> (bucket, last_access)
self._lock = threading.Lock()
def get_bucket(self, key: str) -> TokenBucket:
now = time.monotonic()
with self._lock:
if key in self._buckets:
bucket, _ = self._buckets[key]
self._buckets[key] = (bucket, now)
else:
bucket = TokenBucket(self.rate, self.capacity)
self._buckets[key] = (bucket, now)
# Limpa buckets expirados (best-effort)
expired = [k for k, (_, t) in self._buckets.items()
if now - t > self.ttl]
for k in expired:
del self._buckets[k]
return bucket
@property
def active_buckets(self) -> int:
with self._lock:
return len(self._buckets)
# Uso:
registry = BucketRegistry(rate=10, capacity=20, ttl=1800) # 30 min TTL
bucket = registry.get_bucket("192.168.1.100")
bucket.consume()
O Que Fazer Agora
Copia o código, cola no teu projeto, e testa. Comece com o decorator — é a forma mais fácil de proteger qualquer função sem reescrever nada.
Depois, quando quiser algo mais robusto, migra pro middleware HTTP com o BucketRegistry. A transição é suave porque a interface é a mesma: consume() e wait_and_consume().
Agora me conta: qual foi a última vez que uma automação tua saiu de controle e mandou requisições demais? O que aconteceu? Deixa nos comentários — porque a melhor forma de aprender rate limiting é ter sido a vítima (ou o algoz) de um sistema sem limites.
