Rate Limiter en Python: Token Bucket, Sliding Window y Fixed Window

Tres implementaciones de rate limiting en Python: Token Bucket para permitir ráfagas con llenado continuo, Sliding Window Log para precisión exacta y Fixed Window Counter para la solución más simple. Thread-safe con threading.Lock. Comparativa de cuándo usar cada algoritmo.

				# Rate Limiter en Python: tres implementaciones thread-safe
# Token Bucket · Sliding Window Log · Fixed Window Counter

import threading
import time
from collections import deque


# ?????????????????????????????????????????????????????????????
# Implementación 1: Token Bucket
#
# Una cubeta con capacidad máxima de tokens.
# Se rellena a velocidad constante (rate tokens/segundo).
# Cada petición consume 1 token.
# Permite ráfagas hasta el tamaño de la cubeta.
# ?????????????????????????????????????????????????????????????

class TokenBucket:
    """
    Parámetros:
    - capacity: máximo de tokens (también tamaño máximo de ráfaga)
    - rate: tokens que se añaden por segundo
    """
    def __init__(self, capacity: int, rate: float):
        self.capacity = capacity
        self.rate = rate
        self._tokens = float(capacity)
        self._last_refill = time.monotonic()
        self._lock = threading.Lock()

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

    def allow(self) -> bool:
        with self._lock:
            self._refill()
            if self._tokens >= 1:
                self._tokens -= 1
                return True
            return False


# ?????????????????????????????????????????????????????????????
# Implementación 2: Sliding Window Log
#
# Guarda el timestamp exacto de cada petición en un deque.
# En cada comprobación, descarta los timestamps fuera de la
# ventana y cuenta los que quedan.
# Más preciso que Fixed Window; más costoso en memoria.
# ?????????????????????????????????????????????????????????????

class SlidingWindowLog:
    """
    Parámetros:
    - limit: máximo de peticiones permitidas en la ventana
    - window: tamaño de la ventana en segundos
    """
    def __init__(self, limit: int, window: float):
        self.limit = limit
        self.window = window
        self._log: deque = deque()
        self._lock = threading.Lock()

    def allow(self) -> bool:
        now = time.monotonic()
        cutoff = now - self.window
        with self._lock:
            # Eliminar peticiones antiguas fuera de la ventana
            while self._log and self._log[0] <= cutoff:
                self._log.popleft()
            if len(self._log) < self.limit:
                self._log.append(now)
                return True
            return False


# ?????????????????????????????????????????????????????????????
# Implementación 3: Fixed Window Counter
#
# La más simple: divide el tiempo en ventanas fijas
# y cuenta peticiones por ventana.
# Problema: en el borde entre dos ventanas puede permitir
# hasta el doble del límite en un intervalo corto.
# ?????????????????????????????????????????????????????????????

class FixedWindowCounter:
    """
    Parámetros:
    - limit: máximo de peticiones por ventana
    - window: duración de la ventana en segundos
    """
    def __init__(self, limit: int, window: float):
        self.limit = limit
        self.window = window
        self._count = 0
        self._window_start = time.monotonic()
        self._lock = threading.Lock()

    def allow(self) -> bool:
        now = time.monotonic()
        with self._lock:
            if now - self._window_start >= self.window:
                # Nueva ventana: reiniciar contador
                self._count = 0
                self._window_start = now
            if self._count < self.limit:
                self._count += 1
                return True
            return False


# ?????????????????????????????????????????????????????????????
# Demo: simulación de peticiones
# ?????????????????????????????????????????????????????????????

def simulate(name: str, limiter, requests: int, interval: float) -> None:
    """Envía `requests` peticiones separadas `interval` segundos y muestra el resultado."""
    print(f"n{'?'*50}")
    print(f"{name}")
    print(f"  {requests} peticiones cada {interval*1000:.0f}ms")
    print(f"{'?'*50}")

    allowed = blocked = 0
    for i in range(requests):
        if limiter.allow():
            allowed += 1
            print(f"  [{i+1:2d}] OK")
        else:
            blocked += 1
            print(f"  [{i+1:2d}] BLOQUEADA")
        if i < requests - 1:
            time.sleep(interval)

    print(f"  ? {allowed} permitidas, {blocked} bloqueadas")


if __name__ == '__main__':
    print("=== Rate Limiter: comparativa de algoritmos ===")

    # Token Bucket: capacidad 5, recarga 2 tokens/s
    # Ráfaga inicial de 5, luego 1 cada 500ms
    bucket = TokenBucket(capacity=5, rate=2.0)
    simulate("Token Bucket (cap=5, rate=2/s)", bucket, requests=10, interval=0.1)

    # Sliding Window: 5 peticiones por ventana de 1 segundo
    sw = SlidingWindowLog(limit=5, window=1.0)
    simulate("Sliding Window Log (5 req/s)", sw, requests=10, interval=0.1)

    # Fixed Window: 5 peticiones por ventana de 1 segundo
    fw = FixedWindowCounter(limit=5, window=1.0)
    simulate("Fixed Window Counter (5 req/s)", fw, requests=10, interval=0.1)

    print("n=== Cuándo usar cada uno ===")
    print("""
  Token Bucket:
    + Permite ráfagas controladas (útil para tráfico en lotes)
    + Fácil de ajustar con capacity y rate independientes
    - No da una tasa exacta por ventana de tiempo

  Sliding Window Log:
    + La más precisa: distribuye las peticiones uniformemente
    + No tiene el problema del borde de ventana
    - Más uso de memoria (guarda cada timestamp)

  Fixed Window Counter:
    + La más simple de implementar y entender
    + Mínimo uso de memoria (solo un contador + timestamp)
    - Puede permitir el doble del límite en el borde entre ventanas
""")