Mini-Tutorial: Optimiza tus APIs con Caché
Nivel: Intermedio
Requisitos: Haber completado el curso de Django REST Framework de Platzi
Tiempo: 15-20 minutos
🎯 ¿Por qué es valioso?
Cuando construyes APIs con Django REST Framework, eventualmente te enfrentarás a problemas de rendimiento. Queries pesadas, cálculos complejos o endpoints que se consultan constantemente pueden ralentizar tu aplicación. El caché con Redis es la solución profesional que usan empresas como Instagram, Twitter y Airbnb para mantener sus APIs rápidas.
Lo que aprenderás:
- Implementar caché de forma elegante usando decoradores personalizados
- Invalidar caché automáticamente cuando los datos cambian
- Optimizar endpoints específicos sin modificar toda tu arquitectura
- Un patrón reutilizable que puedes aplicar en cualquier proyecto
📦 Configuración Inicial
1. Instala Redis y el cliente Python
# Instalar Redis (Ubuntu/Debian)
sudo apt update
sudo apt install redis-server
# Instalar cliente Python
pip install redis django-redis
2. Configura Django para usar Redis
En tu settings.py:
# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
},
'KEY_PREFIX': 'drf_cache', # Prefijo para tus keys
'TIMEOUT': 300, # 5 minutos por defecto
}
}
🎨 El Patrón: Decorador Inteligente de Caché
Este es el hack profesional que vas a implementar. Crea un archivo cache_utils.py:
# api/cache_utils.py
from functools import wraps
from django.core.cache import cache
from django.utils.encoding import force_str
from rest_framework.response import Response
import hashlib
import json
def cache_response(timeout=300, key_prefix='api'):
"""
Decorador que cachea las respuestas de tus endpoints.
Args:
timeout: Tiempo de expiración en segundos (default: 5 minutos)
key_prefix: Prefijo para identificar tus keys
Ejemplo:
@cache_response(timeout=600, key_prefix='patients')
def list(self, request):
# Tu código aquí
"""
def decorator(func):
@wraps(func)
def wrapper(self, request, *args, **kwargs):
# Genera una key única basada en la URL, query params y método
cache_key = generate_cache_key(
request=request,
prefix=key_prefix,
view_name=func.__name__,
args=args,
kwargs=kwargs
)
# Intenta obtener del caché
cached_response = cache.get(cache_key)
if cached_response is not None:
# ¡Hit! Retorna desde caché
return Response(cached_response)
# Miss: ejecuta la función original
response = func(self, request, *args, **kwargs)
# Solo cachea respuestas exitosas
if response.status_code == 200:
cache.set(cache_key, response.data, timeout)
return response
return wrapper
return decorator
def generate_cache_key(request, prefix, view_name, args, kwargs):
"""
Genera una key única considerando todos los parámetros relevantes.
"""
# Incluye query params, método HTTP y parámetros de la URL
query_string = request.GET.urlencode()
method = request.method
path_params = '_'.join(str(v) for v in kwargs.values())
# Crea un string único
key_parts = [
prefix,
view_name,
method,
path_params,
query_string
]
key_string = '_'.join(filter(None, key_parts))
# Hash para mantener las keys cortas
key_hash = hashlib.md5(key_string.encode()).hexdigest()
return f"{prefix}:{view_name}:{key_hash}"
def invalidate_cache_pattern(pattern):
"""
Invalida todas las keys que coincidan con un patrón.
Ejemplo:
invalidate_cache_pattern('patients:*')
"""
cache_keys = cache.keys(f"{pattern}")
if cache_keys:
cache.delete_many(cache_keys)
💡 Uso en tus ViewSets
Ahora viene la magia. Aplica el decorador a tus endpoints:
# api/views.py
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import Patient
from .serializers import PatientSerializer
from .cache_utils import cache_response, invalidate_cache_pattern
class PatientViewSet(viewsets.ModelViewSet):
queryset = Patient.objects.all()
serializer_class = PatientSerializer
# 🎯 Cachea el listado por 10 minutos
@cache_response(timeout=600, key_prefix='patients')
def list(self, request):
"""
Este endpoint ahora está optimizado con caché.
La primera llamada tarda lo normal, las siguientes son instantáneas.
"""
queryset = self.get_queryset()
serializer = self.get_serializer(queryset, many=True)
return Response(serializer.data)
# 🎯 Cachea el detalle individual
@cache_response(timeout=300, key_prefix='patients')
def retrieve(self, request, pk=None):
instance = self.get_object()
serializer = self.get_serializer(instance)
return Response(serializer.data)
# ⚡ Invalida el caché cuando se crea un paciente
def create(self, request):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
# Invalida el caché del listado
invalidate_cache_pattern('patients:list:*')
return Response(serializer.data, status=status.HTTP_201_CREATED)
# ⚡ Invalida el caché cuando se actualiza
def update(self, request, pk=None):
instance = self.get_object()
serializer = self.get_serializer(instance, data=request.data)
serializer.is_valid(raise_exception=True)
serializer.save()
# Invalida tanto el listado como el detalle específico
invalidate_cache_pattern('patients:list:*')
invalidate_cache_pattern(f'patients:retrieve:*')
return Response(serializer.data)
# ⚡ Invalida el caché cuando se elimina
def destroy(self, request, pk=None):
instance = self.get_object()
instance.delete()
invalidate_cache_pattern('patients:*')
return Response(status=status.HTTP_204_NO_CONTENT)
# 🎯 Ejemplo con custom action
@action(detail=False, methods=['get'])
@cache_response(timeout=1800, key_prefix='patients_stats')
def statistics(self, request):
"""
Endpoint con cálculos pesados que se beneficia del caché.
"""
total_patients = Patient.objects.count()
# Imagina queries más complejas aquí...
data = {
'total': total_patients,
'cached_at': 'timestamp',
}
return Response(data)
🔥 Ventajas de este Patrón
1. Selectivo y Granular
No cacheas todo tu proyecto, solo los endpoints que lo necesitan.
2. Invalidación Inteligente
Cuando los datos cambian (POST, PUT, DELETE), el caché se invalida automáticamente.
3. Keys Únicas Automáticas
El sistema considera query params, lo que significa que:
/api/patients//api/patients/?search=Juan/api/patients/?ordering=-created_at
Tienen cachés independientes. ¡Perfecto para filtros y búsquedas!
4. Fácil de Mantener
Un simple decorador. No contaminas tu lógica de negocio.
📊 Midiendo el Impacto
Agrega un middleware para ver los hits/misses de caché:
# api/middleware.py
from django.utils.deprecation import MiddlewareMixin
import time
class CacheMetricsMiddleware(MiddlewareMixin):
def process_request(self, request):
request._cache_start_time = time.time()
def process_response(self, request, response):
if hasattr(request, '_cache_start_time'):
duration = time.time() - request._cache_start_time
# Agrega headers para debugging
response['X-Response-Time'] = f"{duration:.3f}s"
return response
Agrégalo a settings.py:
MIDDLEWARE = [
# ... otros middlewares
'api.middleware.CacheMetricsMiddleware',
]
Ahora verás en los headers de respuesta cuánto tardó cada request.
🧪 Testing
No olvides probar tu caché:
# api/tests.py
from django.test import TestCase
from django.core.cache import cache
from rest_framework.test import APIClient
from .models import Patient
class CacheTestCase(TestCase):
def setUp(self):
self.client = APIClient()
cache.clear() # Limpia el caché antes de cada test
def test_list_endpoint_caching(self):
"""Verifica que el listado se cachee correctamente"""
# Primera llamada: debe hacer query a la BD
response1 = self.client.get('/api/patients/')
self.assertEqual(response1.status_code, 200)
# Segunda llamada: debe venir del caché (más rápida)
response2 = self.client.get('/api/patients/')
self.assertEqual(response2.status_code, 200)
self.assertEqual(response1.data, response2.data)
def test_cache_invalidation_on_create(self):
"""Verifica que crear un paciente invalide el caché del listado"""
# Cachea el listado
self.client.get('/api/patients/')
# Crea un nuevo paciente
self.client.post('/api/patients/', {'name': 'Test', 'age': 30})
# El caché debe estar invalidado, así que esta llamada
# debería reflejar el nuevo paciente
response = self.client.get('/api/patients/')
self.assertEqual(len(response.data), 1)
🎓 Nivel Siguiente: Caché con TTL Dinámico
Un hack más avanzado - ajusta el timeout según la hora del día:
from datetime import datetime
def dynamic_cache_timeout():
"""
Caché más corto durante horas pico, más largo en la madrugada.
"""
hour = datetime.now().hour
# 8am - 10pm: caché corto (mucha actividad)
if 8 <= hour <= 22:
return 180 # 3 minutos
else:
# Madrugada: caché largo (poca actividad)
return 1800 # 30 minutos
# Uso:
@cache_response(timeout=dynamic_cache_timeout(), key_prefix='patients')
def list(self, request):
# ...
🚀 Resultado Final
Con esta implementación:
✅ Reducirás la carga de tu base de datos en un 70-90%
✅ Tus endpoints más consultados responderán en <10ms
✅ Mantendrás tu código limpio y mantenible
✅ Aprenderás un patrón usado en producción por empresas grandes
💎 Tips Pro
- Monitorea tu Redis: Usa
redis-cli monitorpara ver las operaciones en tiempo real - No cachees todo: Endpoints de escritura intensiva o datos muy dinámicos no son buenos candidatos
- Considera caché de fragmentos: Para serializers pesados, cachea partes específicas
- Combina con throttling: El caché ayuda, pero throttling protege de abuso
📚 Recursos Adicionales
🎉 Conclusión
Has aprendido un patrón de arquitectura real que:
- Complementa lo aprendido en el curso de Platzi (serializers, viewsets, custom actions)
- Agrega valor inmediato a tus proyectos
- Es reutilizable en cualquier API de Django REST Framework
- Te diferencia como desarrollador que piensa en rendimiento
¡Ahora comparte este conocimiento con la comunidad! 🚀
Por [@tu_usuario]
Basado en el curso de Django REST Framework de Platzi
Curso de Django Rest Framework
COMPARTE ESTE ARTÍCULO Y MUESTRA LO QUE APRENDISTE
0 Comentarios
para escribir tu comentario
