SolveConPython

Python Reto #26 — Rate limiter (Token Bucket simple)

Nivel: Intermedio+
Tema: Control de tasa (rate limiting), tiempo, estado, diseño defensivo, tests con pytest
Objetivo: Implementar un rate limiter estilo Token Bucket para limitar cuántas operaciones se permiten por unidad de tiempo.

Concepto (rápido)

Un Token Bucket tiene:

  • una capacidad máxima de tokens,
  • una tasa de recarga (tokens por segundo),
  • cada operación consume 1 token,
  • si no hay tokens disponibles, la operación se rechaza.

Enunciado

Crea una clase TokenBucketRateLimiter con:

Constructor

TokenBucketRateLimiter(capacidad, tokens_por_segundo, ahora_fn=None)

Reglas:

  1. capacidad debe ser int > 0 (si no: TypeError o ValueError)
  2. tokens_por_segundo debe ser float o int > 0 (si no: TypeError o ValueError)
  3. ahora_fn (opcional) debe ser callable si se proporciona.
    • Si no se proporciona, usa time.time.
    • Esto permite tests deterministas.

Método

allow() -> bool

Debe:

  1. “Recargar” tokens según el tiempo transcurrido desde la última actualización:
    • tokens += (tiempo_transcurrido * tokens_por_segundo)
    • nunca superar capacidad
  2. Si hay al menos 1 token disponible:
    • consumir 1 token,
    • devolver True
  3. Si no:
    • devolver False
  4. Debe funcionar correctamente aunque se llame muchas veces seguidas.

Ejemplos

rl = TokenBucketRateLimiter(capacidad=3, tokens_por_segundo=1)

rl.allow() # True (2 tokens quedan)
rl.allow() # True (1 token queda)
rl.allow() # True (0 tokens quedan)
rl.allow() # False (sin tokens)

# tras ~1 segundo, debería recargar ~1 token

Pistas

  1. Guarda:
    • self.tokens como float (para recargas fraccionales),
    • self.ultimo_tiempo
  2. En allow():
    • calcula delta = ahora - ultimo_tiempo,
    • recarga,
    • actualiza ultimo_tiempo,
    • decide si permitir.
  3. Para tests, inyecta ahora_fn que tú puedas controlar.

Solución explicada (paso a paso)

  1. Al crear el limiter:
    • inicia tokens = capacidad
    • ultimo_tiempo = ahora_fn()
  2. En cada allow():
    • calcula cuánto tiempo pasó,
    • recarga tokens proporcionalmente,
    • limita tokens a capacidad,
    • si tokens >= 1: consume y permite,
    • si no: rechaza.

Ejecuta:

  • pytest -q

Variantes para subir de nivel (opcional)

  1. Coste por operación: allow(cost=1) consumiendo más de 1 token
  2. Multiple buckets por “cliente” (diccionario de buckets por API key)
  3. Devolver tiempo estimado de espera cuando allow() es False
  4. Persistencia (guardar estado entre ejecuciones)

Lo que aprendiste

  • Modelo Token Bucket y por qué se usa en APIs
  • Cómo manejar tiempo y estado de forma segura
  • Tests deterministas inyectando un reloj
  • Validación defensiva de parámetros

Accede al código completo y a los tests en GitHub para ejecutar y modificar la solución localmente.

Siguiente reto recomendado: Reto #27 — Retry con backoff exponencial (sin librerías) para patrones reales de resiliencia en redes y automatización.