Throttling y rate limiting en DRF

Por qué limitar peticiones

Una API sin throttling está expuesta a:

• Abuso intencionado: scrapers que consumen datos, ataques de fuerza bruta a endpoints de login.
• Abuso accidental: un cliente mal programado con bucle infinito de peticiones.
• Carga desequilibrada: un tenant consume el 90% de los recursos y degrada la experiencia de los demás.
• Costes descontrolados: si pagas por peticiones (API externa, compute, BBDD), un abuso dispara la factura.

El throttling (limitar peticiones por periodo) es una defensa imprescindible. DRF lo incorpora con clases reutilizables y un mecanismo sencillo de personalización.

Throttling básico: AnonRateThrottle y UserRateThrottle

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'user': '1000/day',
    },
}

• AnonRateThrottle: aplica a peticiones sin autenticar. Throttling por IP.
• UserRateThrottle: aplica a peticiones autenticadas. Throttling por user ID.

El formato del rate es N/periodo donde periodo ∈ {second, minute, hour, day}.

Cuando un cliente excede el límite, DRF devuelve 429 Too Many Requests con header Retry-After.

Throttling por endpoint con ScopedRateThrottle

A veces quieres límites distintos por operación. Por ejemplo, el login debe ser más estricto que un listado:

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.ScopedRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'login': '5/minute',
        'search': '60/minute',
        'upload': '10/hour',
    },
}

# views.py
class LoginView(APIView):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = 'login'

    def post(self, request):
        ...

class BuscarProductosView(APIView):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = 'search'

    def get(self, request):
        ...

class SubirArchivoView(APIView):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = 'upload'

    def post(self, request):
        ...

Cada scope lleva su propio contador separado.

Throttling combinado: ráfagas + sostenido

Un patrón clásico: permitir ráfagas cortas (p. ej. 60/min) pero también un límite sostenido (1000/día). DRF lo soporta apilando dos clases:

from rest_framework.throttling import UserRateThrottle

class BurstUserRateThrottle(UserRateThrottle):
    scope = 'burst'

class SustainedUserRateThrottle(UserRateThrottle):
    scope = 'sustained'

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'api.throttles.BurstUserRateThrottle',
        'api.throttles.SustainedUserRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'burst': '60/minute',
        'sustained': '1000/day',
    },
}

Ambas se aplican simultáneamente: el cliente puede hacer hasta 60 peticiones en un minuto, siempre que no supere 1000 al día.

Throttling por API key (escenario B2B)

Cuando cada tenant tiene una API key y paga por plan (plan básico 10k/día, plan pro 100k/día):

# throttles.py
from rest_framework.throttling import SimpleRateThrottle

class ApiKeyRateThrottle(SimpleRateThrottle):
    scope = 'apikey'

    def get_cache_key(self, request, view):
        # Recupera la api key del header
        api_key = request.headers.get('X-API-Key')
        if not api_key:
            return None  # sin api key, no aplica throttling (o deja que otro throttle lo gestione)
        return f'throttle_apikey_{api_key}'

    def get_rate(self):
        # Rate distinto segun el plan asociado a la API key
        api_key = self.request.headers.get('X-API-Key') if hasattr(self, 'request') else None
        if api_key:
            plan = ApiKey.objects.get(key=api_key).plan
            return plan.rate  # ej "100000/day"
        return '10/minute'

Usar con:

class ProductoViewSet(viewsets.ModelViewSet):
    throttle_classes = [ApiKeyRateThrottle]

Backend de cache para throttling

Por defecto, DRF usa el cache de Django (que en memoria local). En producción con varios workers, el contador debe estar en un backend compartido. Usa Redis:

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
    },
}

Sin esto, cada worker tiene su propio contador y el cliente podría superar N x rate (N = workers). Con Redis todos los workers comparten el contador.

Headers informativos al cliente

Aunque DRF añade Retry-After, no envía headers estándar como X-RateLimit-Limit. Un middleware que los añade:

# middleware.py
class ThrottleHeadersMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        response = self.get_response(request)

        throttles = getattr(request, '_throttles_checked', [])
        if throttles:
            throttle = throttles[0]
            wait = throttle.wait()
            response['X-RateLimit-Limit'] = str(throttle.num_requests)
            response['X-RateLimit-Window'] = str(throttle.duration)
            if wait:
                response['X-RateLimit-Reset'] = str(int(time.time() + wait))

        return response

Con esto, los clientes bien educados pueden parar antes de recibir 429.

Excluir rutas del throttling

Algunas rutas (health checks, documentación) no deben contar para el throttling:

class HealthView(APIView):
    throttle_classes = []  # sin throttling
    permission_classes = [AllowAny]

    def get(self, request):
        return Response({'status': 'ok'})

Patrones habituales

Login estricto: 5 intentos por 15 minutos

Evita brute-force:

class LoginThrottle(SimpleRateThrottle):
    scope = 'login'
    rate = '5/15min'

    def get_cache_key(self, request, view):
        ident = self.get_ident(request)  # IP por defecto
        return f'throttle_login_{ident}'

Después de 5 intentos, devuelve 429 durante 15 minutos.

Bulk operations: rate más estricto

Un endpoint que permite crear 100 objetos en una petición debe tener throttling más estricto que uno individual. Con Scoped:

class BulkCreateView(APIView):
    throttle_classes = [ScopedRateThrottle]
    throttle_scope = 'bulk'  # en settings: 'bulk': '10/hour'

Anónimos: distinguir crawlers amigos

Los bots de Google, Bing, etc. indican su user agent. Puedes permitirles un rate mayor:

class FriendlyBotThrottle(AnonRateThrottle):
    KNOWN_BOTS = ['Googlebot', 'Bingbot', 'DuckDuckBot']

    def get_cache_key(self, request, view):
        ua = request.headers.get('User-Agent', '')
        if any(bot in ua for bot in self.KNOWN_BOTS):
            self.rate = '10000/day'  # rate elevado
            self.num_requests, self.duration = self.parse_rate(self.rate)
        return super().get_cache_key(request, view)

Combinar throttling con django-ratelimit

Para casos más complejos (ej. limitar por IP + usuario + ruta simultáneamente), la librería django-ratelimit ofrece una DSL más expresiva que se integra con DRF vía decoradores.

from ratelimit.decorators import ratelimit

@ratelimit(key='ip', rate='100/h', block=True)
@ratelimit(key='user', rate='1000/h', block=True)
@api_view(['POST'])
def mi_endpoint(request):
    ...

Monitorizar y alertar

• Log cada 429 con cliente + endpoint + IP para detectar abusos.
• Dashboard en Grafana con tasa de 429 por endpoint.
• Alerta si 429 > 5% del tráfico: indica que los rates están mal configurados o hay un ataque.

Checklist en producción

• [ ] AnonRateThrottle activo para endpoints públicos.
• [ ] UserRateThrottle activo para autenticados.
• [ ] Redis como cache backend para que todos los workers compartan contador.
• [ ] Login con rate estricto (5/15min o similar).
• [ ] Endpoints caros (search, bulk, upload) con ScopedRateThrottle propio.
• [ ] Headers X-RateLimit-* enviados al cliente.
• [ ] Exclusiones documentadas (health check, docs).
• [ ] Monitorización: tasa de 429 < 5% en tráfico normal.