Implementar Redis con Fast-API para añadir cache

Tiempo de lectura: 3 minutos

Redis es una de las bases de datos en memoria más rápidas y populares, diseñada para manejar datos con una latencia mínima. Su capacidad para almacenar datos temporalmente y recuperarlos rápidamente la hace ideal para escenarios como el almacenamiento en caché, la gestión de sesiones, la clasificación de datos en tiempo real y mucho más.

Por otro lado, FastAPI es un framework moderno y rápido para crear APIs en Python, conocido por su eficiencia y facilidad de uso, gracias a la integración nativa de async y type hints. Juntos, Redis y FastAPI forman una combinación poderosa para aplicaciones web y microservicios que necesitan alto rendimiento.

¿Por qué necesitas Redis en una aplicación web?

Redis brilla especialmente cuando trabajas con datos que cambian frecuentemente y necesitas acceso rápido. Algunos casos de uso comunes en una aplicación web incluyen:

• Almacenamiento en caché: Para evitar realizar cálculos o consultas repetitivas a bases de datos lentas.
  Ejemplo: Caché de datos como resultados de búsquedas o información estática.
• Gestión de sesiones: Para manejar información del usuario de manera eficiente y rápida.
  Ejemplo: Almacenar tokens de autenticación o datos temporales de usuarios.
• Contadores y rankings en tiempo real: Perfecto para aplicaciones de análisis o juegos en línea.
• Colas de mensajes: Redis puede funcionar como un sistema de mensajería ligera para eventos en tiempo real.

1. Instalar las dependencias necesarias

Primero, instala las bibliotecas necesarias:

pip install redis fastapi-cache2

Además, asegúrate de tener Redis instalado y funcionando en tu sistema. Si no lo tienes, puedes instalarlo o usar Docker:

docker run -d --name redis -p 6379:6379 redis

2. Configurar FastAPI con Redis

Crea un proyecto FastAPI con configuración para la caché.

Código completo:

from fastapi import FastAPI
from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from redis import asyncio as aioredis
import time

app = FastAPI()

# Inicializa la conexión con Redis en el arranque
@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost")  # Cambia localhost si tu Redis está en otro host
    FastAPICache.init(RedisBackend(redis), prefix="fastapi-cache")

# Endpoint con caché usando FastAPI-Cache
@app.get("/data/")
@cache(expire=60)
async def get_data(tipo: str, dia: str, idioma: str):
    # Simula la generación de datos
    data = {
        "tipo": tipo,
        "dia": dia,
        "idioma": idioma,
        "timestamp": time.time()  # Marca temporal para diferenciar
    }
    return {"source": "generated", "data": data}

3. Cómo funciona el código

1. Conexión a Redis:
   • En el evento startup, se inicializa una conexión asíncrona a Redis utilizando aioredis.
   • FastAPICache.init configura Redis como el backend para la caché.
2. Caché por endpoint:
   • En el endpoint /data/, el decorador @cache(expire=60) almacena los datos generados por la función durante 60 segundos.
   • Si se realiza la misma solicitud con los mismos parámetros dentro de los 60 segundos, los datos serán servidos desde Redis, sin volver a ejecutar la lógica de generación.

4. Probar la aplicación

Ejecuta la aplicación:

uvicorn main:app --reload

Haz solicitudes al endpoint y observa que las respuestas posteriores dentro del tiempo de vida (TTL) serán más rápidas:

1. Primera llamada: Generará datos nuevos. curl "http://127.0.0.1:8000/data/?tipo=x&dia=y&idioma=e" Respuesta: { "source": "generated", "data": { "tipo": "x", "dia": "y", "idioma": "e", "timestamp": 1696352000.123456 } }
2. Segunda llamada (dentro de 300 segundos): Los datos se servirán desde la caché. curl "http://127.0.0.1:8000/data/?tipo=x&dia=y&idioma=e" Respuesta: { "source": "cache", "data": { "tipo": "x", "dia": "y", "idioma": "e", "timestamp": 1696352000.123456 } }

5. Opciones adicionales

• Actualizar datos en caché: Si los datos cambian frecuentemente, puedes invalidar manualmente la caché usando FastAPICache.clear() o implementando una estrategia para invalidar claves específicas.
• Configurar Redis remoto: Si Redis no está en tu localhost, reemplaza redis://localhost por la URL o IP de tu servidor Redis: redis = aioredis.from_url("redis://<IP_DEL_SERVIDOR>:6379")
• Prefijos personalizados: Cambia el prefix="fastapi-cache" para diferenciar cachés si utilizas múltiples aplicaciones o módulos en Redis.