Rate limiting e throttling de API - controle de requisições em servidor Python com SQLite

Rate Limiter Local com Python e SQLite: O Freio de Mão Que Sua Automação Precisa (Antes do 429)

PorOlivetto

Tem uma automação que consome API externa — weather, CRM, gateway de pagamento, sei lá — e de repente ela para de funcionar. Sem explicação. Você verifica os logs e lá está: 429 Too Many Requests.

A primeira reação é sempre a mesma: “deve ser problema deles.” Spoiler: não é. O problema é que sua automação dispara requisições como se o endpoint fosse infinito. E não é. Toda API séria tem limite, e ignorar isso é como acelerar em fila de pedágio — funciona até a multa chegar.

O que eu fiz — e o que vou te mostrar aqui — é construir um rate limiter local com Python e SQLite. Nada de Redis. Nada de serviço gerenciado. Uma classe de 180 linhas que roda no mesmo processo da sua automação, conta requisições em uma tabela SQLite, e bloqueia antes que o 429 apareça.

Se você tem automações que consomem API e quer parar de ser pego de surpresa por throttling, esse post é seu.

Rate limiting e throttling de API - controle de requisições em servidor Python com SQLite
Rate limiting local: o freio de mão que sua automação precisa antes de cair no abismo do 429.

O que é rate limiting (e por que sua automação precisa de um)

Rate limiting é o mecanismo que limita quantas requisições um cliente pode fazer a uma API dentro de uma janela de tempo. Pode ser 100 requisições por minuto, 1.000 por hora, ou qualquer coisa que o provedor decidir.

Quando você ultrapassa esse limite, a API retorna HTTP 429. E aqui está o problema prático: sua automação não espera. Ela continua enviando, acumula erros, e dependendo do sistema, pode até perder dados no processo.

Um rate limiter local resolve isso com uma premissa simples: antes de enviar, verifica se pode enviar. Se não pode, espera. Quando pode, dispara.

Por que local? Por que não Redis?

Redis é ótimo. Sério. Mas se sua automação roda em um VPS de R$30/mês ou num Raspberry Pi na sua estante, adicionar Redis é overkill. SQLite já está no Python. Já está no seu sistema. E para rate limiting local, ele é mais que suficiente.

A diferença prática:

  • Redis: processo extra, configuração, monitoramento, mais um ponto de falha
  • SQLite: arquivo, zero configuração, embedded no processo, backup trivial (copia o arquivo)

Arquitetura do Rate Limiter

O sistema funciona assim:

  1. Tabela de registro: cada requisição é logada com timestamp e identificador do endpoint
  2. Consulta de janela: antes de enviar, conta quantas requisições foram feitas na janela (ex: último minuto)
  3. Decisão: se abaixo do limite, envia. Se acima, calcula quanto esperar e dorme
  4. Cleanup: registros antigos são removidos para manter a tabela leve

É simples porque rate limiting não precisa ser complexo. Precisa ser confiável.

Controle de throttling de API com fila de requisições SQLite
Fila de requisições: requisições entram, o rate limiter decide, só as autorizadas passam.

Implementação Completa

1. A classe base — RateLimiter

Aqui está o esqueleto funcional. Copia, cola, adapta:

import sqlite3
import time
from contextlib import contextmanager
from dataclasses import dataclass


@dataclass
class RateLimitConfig:
    max_requests: int      # Máximo de requisições na janela
    window_seconds: float  # Tamanho da janela em segundos
    endpoint: str = "default"


class RateLimiter:
    def __init__(self, db_path: str = "rate_limiter.db"):
        self.db_path = db_path
        self._init_db()

    def _get_conn(self):
        conn = sqlite3.connect(self.db_path)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA busy_timeout=5000")
        return conn

    def _init_db(self):
        with self._get_conn() as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS request_log (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    endpoint TEXT NOT NULL,
                    timestamp REAL NOT NULL
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_endpoint_time
                ON request_log(endpoint, timestamp)
            """)

    @contextmanager
    def _connection(self):
        conn = self._get_conn()
        try:
            yield conn
            conn.commit()
        finally:
            conn.close()

    def _count_requests(self, endpoint: str, window: float) -> int:
        cutoff = time.time() - window
        with self._connection() as conn:
            cursor = conn.execute(
                "SELECT COUNT(*) FROM request_log WHERE endpoint = ? AND timestamp > ?",
                (endpoint, cutoff)
            )
            return cursor.fetchone()[0]

    def _log_request(self, endpoint: str):
        with self._connection() as conn:
            conn.execute(
                "INSERT INTO request_log (endpoint, timestamp) VALUES (?, ?)",
                (endpoint, time.time())
            )

    def _cleanup_old(self, endpoint: str, max_age: float = 86400):
        cutoff = time.time() - max_age
        with self._connection() as conn:
            conn.execute(
                "DELETE FROM request_log WHERE endpoint = ? AND timestamp < ?",
                (endpoint, cutoff)
            )

    def wait_if_needed(self, config: RateLimitConfig) -> float:
        """Espera se necessário. Retorna tempo de espera (0 se imediato)."""
        self._cleanup_old(config.endpoint)
        current = self._count_requests(config.endpoint, config.window_seconds)

        if current >= config.max_requests:
            cutoff = time.time() - config.window_seconds
            with self._connection() as conn:
                cursor = conn.execute(
                    "SELECT MIN(timestamp) FROM request_log WHERE endpoint = ? AND timestamp > ?",
                    (config.endpoint, cutoff)
                )
                oldest = cursor.fetchone()[0]

            if oldest:
                wait_time = (oldest + config.window_seconds) - time.time()
                wait_time = max(wait_time, 0.1)
                time.sleep(wait_time)
                return wait_time

        return 0.0

    def acquire(self, config: RateLimitConfig):
        """Adquire permissão para enviar requisição."""
        self.wait_if_needed(config)
        self._log_request(config.endpoint)

2. O wrapper HTTP — integrando com requests

import requests
from typing import Optional


class RateLimitedHTTP:
    def __init__(self, rate_limiter: RateLimiter):
        self.limiter = rate_limiter
        self.session = requests.Session()

    def get(self, url: str, config: RateLimitConfig, **kwargs) -> requests.Response:
        self.limiter.acquire(config)
        return self.session.get(url, **kwargs)

    def post(self, url: str, config: RateLimitConfig, **kwargs) -> requests.Response:
        self.limiter.acquire(config)
        return self.session.post(url, **kwargs)

3. Uso real — consumindo uma API com limite

# Configuração: 5 requisições por minuto para a API de weather
weather_config = RateLimitConfig(
    max_requests=5,
    window_seconds=60,
    endpoint="api.weather"
)

crm_config = RateLimitConfig(
    max_requests=10,
    window_seconds=60,
    endpoint="api.crm"
)

http = RateLimitedHTTP(RateLimiter())

for cidade in ["São Paulo", "Rio", "Curitiba", "Salvador", "Recife", "Manaus"]:
    resp = http.get(
        f"https://api.weather.com/forecast?city={cidade}",
        config=weather_config,
        timeout=10
    )
    print(f"{cidade}: {resp.status_code}")

Token Bucket: A Alternativa Mais Flexível

O rate limiter de janela fixa acima funciona, mas tem uma limitação: ele é binário. Ou passa, ou espera. O Token Bucket é mais suave — você acumula “tokens” ao longo do tempo e gasta um por requisição. Se não tem token, espera.

A vantagem? Permite bursts controlados. Se ficou 10 minutos sem usar a API, tem 10 minutos de tokens acumulados. Pode disparar várias requisições seguidas até esgotar o bucket.

class TokenBucketLimiter:
    def __init__(self, db_path: str = "token_bucket.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS buckets (
                    endpoint TEXT PRIMARY KEY,
                    tokens REAL NOT NULL,
                    last_refill REAL NOT NULL,
                    max_tokens REAL NOT NULL,
                    refill_rate REAL NOT NULL
                )
            """)

    def _get_bucket(self, endpoint, max_tokens, refill_rate):
        now = time.time()
        with sqlite3.connect(self.db_path) as conn:
            row = conn.execute(
                "SELECT tokens, last_refill, max_tokens, refill_rate FROM buckets WHERE endpoint = ?",
                (endpoint,)
            ).fetchone()

            if not row:
                conn.execute(
                    "INSERT INTO buckets VALUES (?, ?, ?, ?, ?)",
                    (endpoint, max_tokens, now, max_tokens, refill_rate)
                )
                return max_tokens

            tokens, last_refill, max_tok, rate = row
            elapsed = now - last_refill
            new_tokens = min(max_tok, tokens + elapsed * rate)

            conn.execute(
                "UPDATE buckets SET tokens = ?, last_refill = ? WHERE endpoint = ?",
                (new_tokens, now, endpoint)
            )
            return new_tokens

    def acquire(self, endpoint, max_tokens, refill_rate, tokens_needed=1.0):
        current = self._get_bucket(endpoint, max_tokens, refill_rate)

        if current >= tokens_needed:
            new_tokens = current - tokens_needed
            with sqlite3.connect(self.db_path) as conn:
                conn.execute(
                    "UPDATE buckets SET tokens = ? WHERE endpoint = ?",
                    (new_tokens, endpoint)
                )
            return 0.0

        deficit = tokens_needed - current
        wait_time = deficit / refill_rate
        time.sleep(wait_time)

        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE buckets SET tokens = 0, last_refill = ? WHERE endpoint = ?",
                (time.time(), endpoint)
            )
        return wait_time

Múltiplos Endpoints, Um Só Banco

Num sistema real, você consome várias APIs. Cada uma com limite diferente. O rate limiter precisa gerenciar todas simultaneamente.

class MultiEndpointLimiter:
    def __init__(self, db_path="multi_limiter.db"):
        self.limiter = RateLimiter(db_path)
        self.configs = {}

    def register(self, config):
        self.configs[config.endpoint] = config

    def acquire(self, endpoint):
        config = self.configs.get(endpoint)
        if not config:
            raise ValueError(f"Endpoint '{endpoint}' não registrado")
        self.limiter.acquire(config)

    def status(self):
        """Mostra uso atual de cada endpoint."""
        status = {}
        for name, config in self.configs.items():
            count = self.limiter._count_requests(name, config.window_seconds)
            status[name] = {
                "usado": count,
                "limite": config.max_requests,
                "janela": f"{config.window_seconds}s",
                "disponivel": max(0, config.max_requests - count)
            }
        return status


limiter = MultiEndpointLimiter()
limiter.register(RateLimitConfig(5, 60, "api.weather"))
limiter.register(RateLimitConfig(10, 60, "api.crm"))
limiter.register(RateLimitConfig(100, 3600, "api.email"))

print(limiter.status())
Servidor Python com rate limiter ativo - proteção contra sobrecarga de API
Proteção ativa: rate limiter no caminho entre sua automação e a API externa.

Monitoramento: Sabendo Quando Está Perto do Limite

Um rate limiter sem monitoramento é como velocímetro sem ponteiro — funciona, mas você não sabe a velocidade. Aqui está um decorator simples que loga quando você está acima de 80% do limite:

import logging
import functools

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("rate_limiter")


def warn_near_limit(limiter, config):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            count = limiter._count_requests(config.endpoint, config.window_seconds)
            usage_pct = (count / config.max_requests) * 100

            if usage_pct >= 80:
                logger.warning(
                    f"[RATE LIMIT] {config.endpoint}: {count}/{config.max_requests} "
                    f"({usage_pct:.0f}% usado na janela de {config.window_seconds}s)"
                )

            return func(*args, **kwargs)
        return wrapper
    return decorator


@warn_near_limit(limiter.limiter, limiter.configs["api.weather"])
def buscar_clima(cidade):
    return limiter.limiter.acquire(limiter.configs["api.weather"])

Quando o 429 Acontece Mesmo Assim

Sim, pode acontecer. Se outra instância da sua automação (ou outro serviço) consome a mesma API, seu rate limiter local não sabe. A estratégia é dupla:

  1. Rate limiter conservador: configure 70-80% do limite real. A margem cobre consumo concorrente.
  2. Fallback no 429: quando receber 429, extraia o header Retry-After e respeite-o:
def resilient_get(url, config, limiter, session=None):
    session = session or requests.Session()
    max_retries = 3

    for attempt in range(max_retries):
        limiter.acquire(config)
        resp = session.get(url, timeout=10)

        if resp.status_code == 429:
            retry_after = int(resp.headers.get("Retry-After", 60))
            logger.warning(f"429 recebido. Esperando {retry_after}s antes de retry")
            time.sleep(retry_after)
            continue

        resp.raise_for_status()
        return resp

    raise RuntimeError(f"429 após {max_retries} tentativas para {url}")

Resumo das Escolhas

Decisão Por quê
SQLite em vez de Redis Zero dependência extra, já embutido no Python
Janela deslizante Mais preciso que janela fixa — não reseta no limite
Token Bucket como alternativa Permite bursts controlados quando a API ficou ociosa
70-80% do limite real Margem para consumo concorrente de outros serviços
PRAGMA journal_mode=WAL Performance de escrita sem bloquear leituras simultâneas

Links Relacionados

Se esse tipo de controle te interessa, dá uma olhada nos posts anteriores:

A Pergunta Que Fica

Quantas vezes sua automação quebrou por causa de rate limiting que você não viu chegando? Se a resposta for “mais de uma vez”, esse rate limiter é pra você. Se nunca aconteceu, parabéns — mas vai acontecer. E quando acontecer, você já sabe onde voltar.

Qual automação você quer que eu desmonte e reconstrua no próximo post? Manda a ideia. Se for problema real, vira artigo.

Posts Similares