Caching de respuestas con Redis en DRF

Por qué cachear

Una API que resuelve cada petición yendo a la base de datos tiene dos problemas:

1. Latencia: una query SQL típica tarda entre 5 y 100 ms. Si además serializas objetos complejos con relaciones anidadas, la respuesta puede superar los 500 ms.
2. Escalabilidad: cada petición ocupa un worker y una conexión de BBDD. Con tráfico alto se agota el pool.

Cache Redis soluciona ambos: guardar la respuesta serializada en memoria y servirla en 1-2 ms sin tocar la BBDD. Típico: 10x a 100x mejora en endpoints de lectura.

Configuración de Redis como backend

pip install django-redis

# settings.py
CACHES = {
    'default': {
        'BACKEND': 'django.core.cache.backends.redis.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'db': '1',
            'parser_class': 'redis.connection.PythonParser',
            'pool_class': 'redis.BlockingConnectionPool',
        },
        'KEY_PREFIX': 'miapi',
    },
}

# Sesiones y cache compartidos
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'

Con Django 4.2+ está disponible django.core.cache.backends.redis.RedisCache en core, sin librerías externas (aunque django-redis sigue siendo más flexible).

Cache a nivel de vista con cache_page

from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
from rest_framework import viewsets

class CategoriaViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = Categoria.objects.all()
    serializer_class = CategoriaSerializer

    @method_decorator(cache_page(60 * 15))  # 15 minutos
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

cache_page(N) guarda la respuesta completa durante N segundos, clave por URL. Los query params forman parte de la clave, así /categorias/ y /categorias/?page=2 se cachean por separado.

Variar por header

Si la respuesta depende de un header (Accept-Language, Authorization), usa vary_on_headers:

from django.views.decorators.vary import vary_on_headers

@method_decorator([cache_page(60 * 15), vary_on_headers('Accept-Language')])
def list(self, request, *args, **kwargs):
    ...

CacheResponseMixin de drf-extensions

pip install drf-extensions

from rest_framework_extensions.cache.mixins import CacheResponseMixin

class CategoriaViewSet(CacheResponseMixin, viewsets.ReadOnlyModelViewSet):
    queryset = Categoria.objects.all()
    serializer_class = CategoriaSerializer

    list_cache_timeout = 60 * 15     # lista cacheada 15min
    object_cache_timeout = 60 * 60   # retrieve cacheado 1h

Cachea list() y retrieve() automáticamente. Clave basada en URL + user + versión (configurable).

Invalidación cuando los datos cambian

El problema principal del caching: invalidación. Si un usuario modifica una categoría, la cache debe actualizarse o el resto de clientes verá datos viejos.

Opción 1: TTL corto y aceptar cierta obsolescencia

list_cache_timeout = 60  # 1 minuto

Simple, pero los cambios tardan en verse. Útil para datos no críticos.

Opción 2: Invalidar con signals

# signals.py
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from .models import Categoria

@receiver([post_save, post_delete], sender=Categoria)
def invalidar_cache_categorias(sender, **kwargs):
    # Borra todas las claves que empiecen por 'cats_'
    cache.delete_pattern('cats_*')

Requiere django-redis (el cache backend por defecto de Django no soporta delete_pattern).

Opción 3: Cache versioning (recomendada)

Usa una versión global que se incrementa cuando cambian los datos. Todas las claves incluyen esa versión. Invalidar = incrementar versión:

# cache_helpers.py
from django.core.cache import cache

def get_categorias_version():
    v = cache.get('version:categorias')
    if v is None:
        v = 1
        cache.set('version:categorias', v, None)  # sin TTL
    return v

def invalidar_categorias():
    cache.incr('version:categorias')

# signals.py
@receiver([post_save, post_delete], sender=Categoria)
def invalidar(sender, **kwargs):
    invalidar_categorias()

# views.py
class CategoriaViewSet(viewsets.ReadOnlyModelViewSet):
    ...

    def list(self, request, *args, **kwargs):
        version = get_categorias_version()
        cache_key = f'categorias:list:v{version}:{request.query_params.urlencode()}'
        cached = cache.get(cache_key)
        if cached:
            return Response(cached)
        response = super().list(request, *args, **kwargs)
        cache.set(cache_key, response.data, 60 * 60)
        return response

Ventaja: no borras claves, simplemente dejas de usarlas. Redis las elimina con TTL. Es atómico y rápido.

ETag y respuestas 304

Con ETag el cliente envía el hash de la última respuesta. Si no ha cambiado, el servidor responde 304 Not Modified sin body:

from django.views.decorators.http import etag
from django.utils.decorators import method_decorator

def get_etag_categorias(request, *args, **kwargs):
    version = get_categorias_version()
    return f'"v{version}"'

class CategoriaViewSet(viewsets.ReadOnlyModelViewSet):
    @method_decorator(etag(get_etag_categorias))
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

El cliente envía If-None-Match: "v42" y si el ETag coincide, responde 304. Ahorra ancho de banda y procesamiento incluso sin cache de Django.

Cache por usuario

No siempre se puede cachear igual para todos. Si el endpoint devuelve datos específicos del usuario (su carrito, sus pedidos), la clave debe incluir el user ID:

from rest_framework_extensions.cache.decorators import cache_response

class MisPedidosView(generics.ListAPIView):
    serializer_class = PedidoSerializer

    def get_queryset(self):
        return Pedido.objects.filter(usuario=self.request.user)

    @cache_response(
        timeout=60 * 5,
        key_func='object_cache_key_func',  # incluye user.id
    )
    def list(self, request, *args, **kwargs):
        return super().list(request, *args, **kwargs)

drf-extensions tiene helpers para construir la clave de forma personalizada (DefaultKeyConstructor, ModelInstanceKeyBit, UserKeyBit, etc.).

Patrón: read-through cache para queries costosas

Si una query tarda 2 segundos pero apenas cambia, puedes envolver en cache la función:

from django.core.cache import cache

def top_productos_por_categoria():
    key = f'top_productos:v{get_categorias_version()}'
    data = cache.get(key)
    if data is None:
        data = list(
            Producto.objects.values('categoria_id')
            .annotate(total=Sum('ventas__cantidad'))
            .order_by('-total')[:100]
        )
        cache.set(key, data, 60 * 60)  # 1 hora
    return data

Aquí el primer request paga los 2 segundos; los siguientes durante 1 hora (o hasta invalidación) obtienen la respuesta en 2 ms.

Cache warming

Evitar el "thundering herd" cuando la cache expira: justo después del TTL, cientos de requests pegan todos a la vez a la BBDD. Soluciones:

• TTL con jitter: timeout = base + random.randint(-30, 30) para que los expiran escalonados.
• Stale-while-revalidate: devolver la cache vieja mientras una task en background la renueva.
• Pre-warm en el arranque del servidor: llama a los endpoints críticos internamente al iniciar.

Medir el impacto

# middleware.py
import time

class CacheStatsMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        start = time.monotonic()
        response = self.get_response(request)
        duration = time.monotonic() - start
        # Log o emitir metrica
        return response

Dashboards: tasa de cache hits, latencia p50/p95/p99 por endpoint, peticiones ahorradas a BBDD. Redis tiene INFO stats con keyspace_hits / keyspace_misses.

Errores comunes

• Cachear datos del usuario sin incluir user_id en la clave: devuelves los pedidos de Juan a María.
• No invalidar tras escritura: los usuarios siguen viendo datos viejos durante el TTL.
• Cachear endpoints con paginación sin incluir query params: devuelves siempre la página 1.
• Cachear con TTL infinito y nunca invalidar: el cache crece sin freno.
• Olvidar Vary header: CDNs y proxies pueden mezclar respuestas personalizadas.
• Redis down sin fallback: si se cae Redis, el endpoint debe seguir funcionando (sin cache pero sin errores). Usa CACHE_SAFE=True (django-redis).

Checklist

• [ ] Redis configurado como backend con django-redis.
• [ ] Endpoints de lectura pesada con cache_page o CacheResponseMixin.
• [ ] Invalidación con signals + versioning para reflejar cambios rápido.
• [ ] Claves que incluyen user.id donde los datos son específicos.
• [ ] ETag en endpoints que soportan 304.
• [ ] TTL razonable (5-60 min para datos cambiantes, horas para semi-estáticos).
• [ ] Fallback a BBDD si Redis no responde.
• [ ] Métricas de hit rate y latencia en Grafana.