Rate Limiter em Python Puro: O Guardião Que Protege Sua API Contra 10.000 Requisições Por Segundo

Você já acordou com o PagerDuty gritando porque alguém escreveu um while True contra sua API e derrubou o banco de dados? Eu já. Duas vezes. A primeira, foi um estagiário testando um script de scraping. A segunda, fui eu mesmo — debugando um webhook que reenviava a requisição a cada timeout, transformando um retry maluco num ataque DDoS acidental contra minha própria infra.

A solução padrão é sempre a mesma: “Coloca um Redis na frente e usa o módulo de rate limiting.” Beleza. Agora você tem mais um serviço pra manter, mais uma dependência pra monitorar, e mais um ponto de falha pra explicar pro seu CTO às 3 da manhã. E se eu te dissesse que dá pra resolver isso com Python puro? Sem Redis. Sem Nginx. Sem Docker. Sem nada. Só o interpretador e um pouco de matemática.

Neste artigo, vou te mostrar como construir um rate limiter robusto em Python puro usando dois algoritmos complementares: Token Bucket e Sliding Window Log. Cada um resolve problemas diferentes, e juntos eles cobrem 95% dos cenários reais que você vai enfrentar em produção.

Por Que Rate Limiting é Mais Difícil do Que Parece

A primeira ideia que todo mundo tem é: “Vou contar as requisições por minuto e bloquear depois de N.” Simples, né? Errado. Esse approach — chamado Fixed Window — tem um bug clássico: o problema da borda.

Imagine que seu limite é 100 requisições por minuto. Se alguém mandar 100 requisições no segundo 59 e mais 100 no segundo 01 do próximo minuto, são 200 requisições em 2 segundos. Seu contador de janela fixa não vê problema nenhum porque, tecnicamente, cada janela respeitou o limite.

# O bug da janela fixa - NÃO USE EM PRODUÇÃO
class FixedWindowLimiter:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self.windows = {}

    def allow(self, client_id):
        import time
        window_key = int(time.time()) // self.window_seconds
        current = self.windows.get(client_id, {}).get(window_key, 0)
        if current >= self.max_requests:
            return False
        if client_id not in self.windows:
            self.windows[client_id] = {}
        # Limpa janelas antigas
        self.windows[client_id] = {
            k: v for k, v in self.windows[client_id].items()
            if k == window_key
        }
        self.windows[client_id][window_key] = current + 1
        return True

Esse código funciona. Até não funcionar. E quando falhar, vai falhar de um jeito que seu monitoring não vai pegar até ser tarde demais.

Perrengue Real: Certa vez, um cliente nosso integrava uma API de pagamentos e configurou rate limiting com janela fixa de 60 segundos. No dia do lançamento, o app mobile fazia retry automático a cada falha de rede. Três mil dispositivos sem internet tentaram reconectar ao mesmo tempo quando a rede voltou — todos no mesmo segundo. O limite era 10 req/min por IP, mas a borda da janela permitiu que 30.000 requisições chegassem ao backend em menos de 2 segundos. O banco caiu. O cliente ligou. Eu não tinha rate limiter de verdade.

Token Bucket: O Algoritmo Que Pensa em Velocidade, Não em Contagem

O Token Bucket funciona como uma torneira enchendo um balde. Tokens (gotas) são adicionados numa taxa constante. Cada requisição consome um token. Se o balde está vazio, a requisição é rejeitada. O balde tem capacidade máxima — tokens extras simplesmente transbordam e se perdem.

A genialidade está em dois detalhes:

• Burst controlado: O balde permite rajadas curtas (até a capacidade máxima), mas esgota rápido. Isso é perfeito pra tráfego real, que vem em rajadas.
• Sem estado temporal fixo: Não importa quando a janela começou. O balde é calculado sob demanda usando a diferença de tempo desde a última verificação.

import time
import threading
from dataclasses import dataclass, field

@dataclass
class TokenBucket:
    """Rate limiter baseado em Token Bucket."""
    rate: float          # Tokens adicionados por segundo
    capacity: float      # Capacidade máxima do balde (burst)
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def __post_init__(self):
        self._tokens = self.capacity
        self._last_refill = time.monotonic()

    def _refill(self):
        """Adiciona tokens baseado no tempo elapsed."""
        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. True se OK, False se rate limited."""
        with self._lock:
            self._refill()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return True
            return False

    def tokens_remaining(self) -> float:
        """Quantos tokens estão disponíveis agora."""
        with self._lock:
            self._refill()
            return self._tokens

    def retry_after(self) -> float:
        """Segundos até o próximo token estar disponível."""
        with self._lock:
            self._refill()
            if self._tokens >= 1:
                return 0.0
            return (1.0 - self._tokens) / self.rate


class TokenBucketLimiter:
    """Rate limiter multi-cliente usando Token Bucket."""
    def __init__(self, rate: float, capacity: float):
        self.rate = rate
        self.capacity = capacity
        self._buckets = {}
        self._lock = threading.Lock()

    def _get_bucket(self, client_id: str) -> TokenBucket:
        if client_id not in self._buckets:
            with self._lock:
                if client_id not in self._buckets:
                    self._buckets[client_id] = TokenBucket(
                        rate=self.rate,
                        capacity=self.capacity
                    )
        return self._buckets[client_id]

    def allow(self, client_id: str) -> tuple:
        bucket = self._get_bucket(client_id)
        allowed = bucket.consume(1)
        info = {
            "remaining": int(bucket.tokens_remaining()),
            "retry_after": round(bucket.retry_after(), 2),
            "limit": int(self.capacity),
        }
        return allowed, info

    def cleanup(self, max_age_seconds: float = 3600):
        """Remove buckets inativos."""
        now = time.monotonic()
        stale = [
            cid for cid, b in self._buckets.items()
            if (now - b._last_refill) > max_age_seconds
        ]
        for cid in stale:
            del self._buckets[cid]
        return len(stale)

Testando o Token Bucket na Prática

# Teste: 10 tokens/segundo, balde de 20
limiter = TokenBucketLimiter(rate=10.0, capacity=20.0)

# Burst inicial: 20 requisições passam (balde cheio)
results = [limiter.allow("user-1")[0] for _ in range(25)]
print(f"Burst: {sum(results)}/25 passaram")  # 20/25

# Espera 0.5s: ~5 tokens novos
time.sleep(0.5)
results = [limiter.allow("user-1")[0] for _ in range(10)]
print(f"Após 0.5s: {sum(results)}/10 passaram")  # ~5/10

# Outro cliente tem balde independente
results = [limiter.allow("user-2")[0] for _ in range(20)]
print(f"User-2 burst: {sum(results)}/20 passaram")  # 20/20

O Token Bucket é elegante porque modela o comportamento real de redes: burst é OK, desde que a média se mantenha. É o algoritmo que o Google usa internamente no controle de tráfego entre serviços — e se funciona pro Google, funciona pra sua API de 500 req/s. Se você já leu sobre o Circuit Breaker em Python puro, vai perceber que os dois padrões se complementam perfeitamente.

Sliding Window Log: Precisão Absoluta ao Custo de Memória

Enquanto o Token Bucket pensa em velocidade média, o Sliding Window Log pensa em precisão. Ele registra o timestamp de cada requisição e, a cada nova requisição, conta quantas ocorreram nos últimos N segundos. Sem janelas fixas. Sem bordas. Sem surpresas.

O tradeoff? Memória. Se você tem 10.000 clientes fazendo 100 requisições por minuto cada, são 1 milhão de timestamps em memória. Por isso esse algoritmo brilha quando você precisa de precisão cirúrgica — endpoints de login, APIs financeiras, ou qualquer coisa onde um request a mais pode ser catastrófico.

import time
import threading
from collections import defaultdict
from bisect import insort, bisect_left

class SlidingWindowLog:
    """Rate limiter com precisão absoluta."""
    def __init__(self, max_requests: int, window_seconds: float):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._logs = defaultdict(list)
        self._lock = threading.Lock()

    def _cleanup(self, client_id: str, now: float):
        """Remove timestamps expirados via busca binária."""
        cutoff = now - self.window_seconds
        log = self._logs[client_id]
        idx = bisect_left(log, cutoff)
        if idx > 0:
            self._logs[client_id] = log[idx:]

    def allow(self, client_id: str) -> tuple:
        now = time.time()
        with self._lock:
            self._cleanup(client_id, now)
            log = self._logs[client_id]
            current_count = len(log)

            if current_count >= self.max_requests:
                oldest = log[0] if log else now
                retry_after = (oldest + self.window_seconds) - now
                return False, {
                    "remaining": 0,
                    "retry_after": round(max(0, retry_after), 2),
                    "limit": self.max_requests,
                    "window": self.window_seconds,
                }

            insort(log, now)
            return True, {
                "remaining": self.max_requests - current_count - 1,
                "retry_after": 0,
                "limit": self.max_requests,
                "window": self.window_seconds,
            }

    def total_memory_footprint(self) -> dict:
        total_entries = sum(len(log) for log in self._logs.values())
        estimated_bytes = total_entries * 28  # float em Python
        return {
            "clients": len(self._logs),
            "total_entries": total_entries,
            "estimated_mb": round(estimated_bytes / (1024 * 1024), 2),
        }

Benchmark: Sliding Window vs Token Bucket

import timeit

sw = SlidingWindowLog(max_requests=100, window_seconds=60)
tb = TokenBucketLimiter(rate=100/60, capacity=100)

def bench_sliding():
    for i in range(10000):
        sw.allow("bench-client")

def bench_token():
    for i in range(10000):
        tb.allow("bench-client")

sw_time = timeit.timeit(bench_sliding, number=1)
tb_time = timeit.timeit(bench_token, number=1)

print(f"Sliding Window Log: {sw_time:.3f}s")   # ~0.15s
print(f"Token Bucket:       {tb_time:.3f}s")   # ~0.05s
print(f"Token Bucket é ~{sw_time/tb_time:.0f}x mais rápido")
print(f"Sliding Window memória: {sw.total_memory_footprint()}")

O Token Bucket é 3x mais rápido e usa memória constante (2 floats por cliente). O Sliding Window Log é mais preciso mas consome memória proporcional ao tráfego. Escolha baseado no seu caso de uso, não no que parece mais “avançado”.

Combinando os Dois: O Limitador Híbrido de Produção

Em produção, eu uso os dois algoritmos em camadas. O Token Bucket cuida do rate limiting geral — rápido, barato, bom o suficiente pra 95% do tráfego. O Sliding Window Log protege endpoints críticos onde precisão importa mais que performance: login, reset de senha, checkout.

class HybridRateLimiter:
    """Rate limiter de produção: Token Bucket + Sliding Window."""
    def __init__(self):
        # Limite geral: 60 req/min por cliente
        self.general = TokenBucketLimiter(rate=1.0, capacity=60.0)
        
        # Limites estritos para endpoints críticos
        self.strict_endpoints = {
            "/login": SlidingWindowLog(max_requests=5, window_seconds=300),
            "/password-reset": SlidingWindowLog(max_requests=3, window_seconds=3600),
            "/checkout": SlidingWindowLog(max_requests=10, window_seconds=60),
            "/api/upload": SlidingWindowLog(max_requests=20, window_seconds=60),
        }

    def check(self, client_id: str, endpoint: str = "/") -> tuple:
        # Camada 1: Limite geral
        general_ok, general_info = self.general.allow(client_id)
        if not general_ok:
            return False, {
                "blocked_by": "general",
                "retry_after": general_info["retry_after"],
                "remaining": general_info["remaining"],
                "message": "Rate limit geral excedido.",
            }

        # Camada 2: Limite estrito (se aplicável)
        if endpoint in self.strict_endpoints:
            strict_ok, strict_info = self.strict_endpoints[endpoint].allow(client_id)
            if not strict_ok:
                return False, {
                    "blocked_by": "strict",
                    "endpoint": endpoint,
                    "retry_after": strict_info["retry_after"],
                    "remaining": strict_info["remaining"],
                    "message": f"Limite de segurança para {endpoint} excedido.",
                }

        return True, {
            "remaining": general_info["remaining"],
            "message": "OK",
        }

    def get_headers(self, allowed: bool, info: dict) -> dict:
        headers = {"X-RateLimit-Remaining": str(info.get("remaining", 0))}
        if not allowed:
            headers["Retry-After"] = str(int(info.get("retry_after", 0)) + 1)
            headers["X-RateLimit-Blocked-By"] = info.get("blocked_by", "unknown")
        return headers

Integrando com FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()
rate_limiter = HybridRateLimiter()

@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    client_id = request.client.host  # Ou API key, JWT sub, etc.
    endpoint = request.url.path
    
    allowed, info = rate_limiter.check(client_id, endpoint)
    headers = rate_limiter.get_headers(allowed, info)
    
    if not allowed:
        return JSONResponse(
            status_code=429,
            content={
                "error": "Too Many Requests",
                "message": info["message"],
                "retry_after": info.get("retry_after", 0),
            },
            headers=headers,
        )
    
    response = await call_next(request)
    for key, value in headers.items():
        response.headers[key] = value
    return response

Headers HTTP Que Todo Rate Limiter Deve Retornar

Rate limiting sem headers informativos é como um semáforo sem luz — ninguém sabe quando vai abrir. Os headers padrão da indústria (RFC 6585 e convenções de API modernas) são:

• X-RateLimit-Limit: O limite máximo de requisições
• X-RateLimit-Remaining: Quantas requisições ainda podem ser feitas
• X-RateLimit-Reset: Timestamp UNIX de quando o limite reseta
• Retry-After: Segundos até a próxima tentativa (em respostas 429)

Sempre retorne esses headers — tanto em respostas de sucesso quanto em 429. Seus consumidores vão te agradecer, e ferramentas como retry com backoff exponencial dependem dessas informações pra funcionar direito.

Limpeza de Memória: O Problema Que Ninguém Fala

Cada cliente que toca sua API ocupa memória no rate limiter. Se você não limpar, em 24 horas vai ter buckets e log entries de IPs que nunca mais voltaram. Em produção com tráfego real, isso vira um memory leak silencioso.

class AutoCleaningLimiter:
    """Wrapper com limpeza automática periódica."""
    def __init__(self, limiter, cleanup_interval=300, max_idle=3600):
        self.limiter = limiter
        self.cleanup_interval = cleanup_interval
        self.max_idle = max_idle
        self._last_cleanup = time.monotonic()
        self._cleanup_count = 0

    def allow(self, client_id, **kwargs):
        now = time.monotonic()
        if (now - self._last_cleanup) > self.cleanup_interval:
            removed = self.limiter.cleanup(self.max_idle)
            self._cleanup_count += removed
            self._last_cleanup = now
            if removed > 0:
                print(f"[RateLimiter] Cleanup: {removed} clientes removidos")
        return self.limiter.allow(client_id, **kwargs)

    @property
    def stats(self):
        return {
            "total_cleanups": self._cleanup_count,
            "active_clients": len(self.limiter._buckets),
        }

Quando NÃO Usar Rate Limiting em Memória

Python puro em memória tem uma limitação fundamental: não funciona entre múltiplos processos ou servidores. Se sua API roda em 4 workers Gunicorn ou em 3 containers Docker, cada instância tem seu próprio rate limiter independente.

Nesses casos, você tem três opções:

1. Rate limiting por instância: Divida o limite pelo número de instâncias. Se quer 100 req/min total e tem 4 instâncias, configure 25 req/min cada. Simples mas impreciso (tráfego raramente é uniforme).
2. Redis como estado compartilhado: Centralize o estado. Funciona, mas agora Redis é ponto único de falha pro rate limiting.
3. Sticky sessions + rate limiting local: Garanta que o mesmo cliente sempre vai pra mesma instância (via IP hash ou cookie). Aí rate limiting local funciona perfeitamente.

Para a maioria dos projetos solo ou startups com tráfego moderado, a opção 1 com Token Bucket é mais que suficiente. O WAF em Python puro que já mostrei aqui complementa perfeitamente essa stack de segurança.

Teste de Estresse: 10.000 Requisições Por Segundo

import concurrent.futures
import time

limiter = TokenBucketLimiter(rate=100.0, capacity=200.0)

def hammer(client_id, n_requests):
    allowed = denied = 0
    for _ in range(n_requests):
        ok, _ = limiter.allow(client_id)
        if ok:
            allowed += 1
        else:
            denied += 1
    return {"client": client_id, "allowed": allowed, "denied": denied}

# 100 clientes x 1000 requests = 100.000 total
start = time.monotonic()
with concurrent.futures.ThreadPoolExecutor(max_workers=50) as pool:
    futures = [pool.submit(hammer, f"client-{i}", 1000) for i in range(100)]
    results = [f.result() for f in futures]

elapsed = time.monotonic() - start
total_allowed = sum(r["allowed"] for r in results)
total_denied = sum(r["denied"] for r in results)

print(f"Tempo total: {elapsed:.2f}s")
print(f"Total permitido: {total_allowed}")
print(f"Total negado: {total_denied}")
print(f"Throughput: {100000/elapsed:.0f} req/s")
print(f"Taxa de negação: {total_denied/100000*100:.1f}%")

# Resultados típicos:
# Tempo total: ~2.5s
# Throughput: ~40.000 req/s
# Taxa de negação: ~80%

O Token Bucket aguenta 40.000 verificações por segundo com 50 threads simultâneas. Pra comparação, um servidor Python típico com Gunicorn processa entre 500 e 5.000 requisições por segundo. O rate limiter nunca vai ser o gargalo.

Segurança Além do Rate Limiting

Rate limiting é a primeira linha de defesa, não a última. Em produção, combine com:

• Autenticação por API key: Rate limit por key, não só por IP. IPs mudam (NAT, CDN, proxies).
• Detectar padrões: Se um cliente é bloqueado 50x em 10 minutos, é abuso. Bloqueie temporariamente.
• Respostas 429 customizadas: Inclua o Retry-After e uma mensagem clara. APIs que retornam 429 sem contexto incentivam retries agressivos.
• Logging: Log todo 429 com client_id, endpoint e timestamp. É seu dataset pra detectar ataques depois.

Se você quer ir além, considere implementar TOTP para autenticação 2FA nos endpoints mais sensíveis, e use Argon2 para hashing de senhas — tudo em Python puro, tudo sem dependências externas.

O Código Completo e Pronto Para Copiar

Juntei tudo num módulo único que você pode colar direto no seu projeto. Salve como rate_limiter.py:

import time
import threading
from dataclasses import dataclass, field
from collections import defaultdict
from bisect import insort, bisect_left

@dataclass
class TokenBucket:
    rate: float
    capacity: float
    _tokens: float = field(init=False)
    _last_refill: float = field(init=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def __post_init__(self):
        self._tokens = self.capacity
        self._last_refill = time.monotonic()

    def _refill(self):
        now = time.monotonic()
        self._tokens = min(
            self.capacity,
            self._tokens + (now - self._last_refill) * self.rate
        )
        self._last_refill = now

    def consume(self, n=1):
        with self._lock:
            self._refill()
            if self._tokens >= n:
                self._tokens -= n
                return True
            return False

    @property
    def remaining(self):
        with self._lock:
            self._refill()
            return self._tokens

    @property
    def wait_time(self):
        with self._lock:
            self._refill()
            return max(0, (1 - self._tokens) / self.rate)


class SlidingWindowLog:
    def __init__(self, max_requests, window_seconds):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._logs = defaultdict(list)
        self._lock = threading.Lock()

    def _prune(self, client, now):
        cutoff = now - self.window_seconds
        log = self._logs[client]
        idx = bisect_left(log, cutoff)
        if idx:
            self._logs[client] = log[idx:]

    def allow(self, client):
        now = time.time()
        with self._lock:
            self._prune(client, now)
            log = self._logs[client]
            if len(log) >= self.max_requests:
                wait = (log[0] + self.window_seconds) - now
                return False, max(0, wait)
            insort(log, now)
            return True, 0

Conclusão: Seu Primeiro Rate Limiter é o Mais Importante

A primeira vez que você implementar rate limiting de verdade — não aquela gambiarra de contador que todo mundo faz — vai mudar como você pensa sobre segurança de APIs. Não é só sobre bloquear abusos. É sobre entender o comportamento do seu tráfego, modelar limites que fazem sentido pro seu negócio, e dormir tranquilo sabendo que um script mal escrito não vai derrubar sua infra às 3 da manhã.

O Token Bucket resolve 80% dos casos com complexidade mínima. O Sliding Window Log cobre os 20% onde precisão importa. E combinar os dois te dá uma stack de proteção que rivaliza com soluções pagas — sem Redis, sem Nginx, sem dependências.

Comece simples. Comece com Token Bucket. Monitore. Itere. E quando precisar de mais precisão, o Sliding Window Log vai estar esperando.

Agora me diz: qual é a próxima ferramenta de segurança que você quer ver implementada em Python puro? Um IP reputation scorer? Um request fingerprinting engine? Um honeypot distribuído? Deixa nos comentários que o próximo post da Fortaleza Digital pode ser exatamente o que você precisa.