FastAPI Integration Profunda

Descripción

Esta cápsula consolida todo lo aprendido del módulo 4 en el patrón de integración profesional con FastAPI: el scaffolding técnico que toda app de producción debería tener para usar Redis correctamente. Vas a aprender 4 patterns que separan código de juguete de código portfolio-worthy: dependency injection con Depends(get_redis) para que cada endpoint reciba el cliente sin acoplarse a un singleton global, lifespan events que inicializan el pool al startup y lo cierran limpio al shutdown, middleware de caching automático que cachea responses sin que cada endpoint tenga que pensar en eso, y graceful degradation completa con logging estructurado.

Este es el módulo 4 cap 04: la cápsula que ata los hilos. La cápsula 02 mostró Pub/Sub. La 03 mostró redis.asyncio y connection pooling. Aquí los integras dentro de FastAPI con los patterns idiomáticos del framework. Al cerrar la cápsula, vas a tener una app FastAPI con Redis profesionalmente integrado: pool inicializado al arranque, dependency injection en todos los endpoints, middleware que cachea automáticamente endpoints GET, health check que reporta status de Redis, y degradación elegante cuando Redis falla. Es el scaffolding que el módulo 5 (Production Cached API) asume desde el día 1.

No hay nuevos conceptos pesados aquí — son combinaciones de los patterns que ya viste, ensamblados profesionalmente. Pero la diferencia entre saber los conceptos individuales y saber integrarlos cohesivamente es lo que diferencia un developer junior de un senior. Esta cápsula es donde haces ese salto.

El stack final

FastAPI app
    │
    ├── Lifespan events: init_pool() en startup, close_pool() en shutdown
    │
    ├── Dependency injection: Depends(get_redis) en endpoints
    │
    ├── Middleware: caching automático para GET endpoints
    │
    ├── Routes:
    │   ├── /health (verifica Redis up)
    │   ├── /api/* (con cache automático via middleware)
    │   └── /admin/* (admin operations)
    │
    └── Graceful degradation: try/except en cada operación Redis

Patrón 1: Dependency Injection

Por qué DI en lugar de import directo

# ❌ Antipatrón: import directo del singleton
from redis_client import get_redis


@app.get("/products/{id}")
async def get_product(id: int):
    r = get_redis()  # acoplado al singleton global
    # ...

Funciona, pero tiene problemas:

1. Tests difíciles: para mockear Redis, tienes que mockear el módulo entero
2. Acoplamiento: cada endpoint conoce la implementación del singleton
3. No es idiomático en FastAPI: el framework provee DI por una razón

Patrón con Depends

from fastapi import Depends
from redis.asyncio import Redis

from redis_client import get_redis_client


# Dependency: función que retorna el cliente
async def get_redis() -> Redis:
    return get_redis_client()


# Endpoint usa Depends
@app.get("/products/{id}")
async def get_product(id: int, r: Redis = Depends(get_redis)):
    cached = await r.get(f"product:{id}")
    if cached:
        return json.loads(cached)
    # ...

Beneficios reales

1. Tests con mocks limpios:

async def get_redis_mock():
    return MockRedis()


def test_get_product():
    app.dependency_overrides[get_redis] = get_redis_mock
    # test...
    del app.dependency_overrides[get_redis]

2. Inyección con configuración por endpoint:

async def get_redis_with_db(db: int = 0):
    """Diferente DB de Redis para diferentes endpoints."""
    return Redis(host='localhost', port=6379, db=db, decode_responses=True)


@app.get("/cache-data")
async def cache_endpoint(r: Redis = Depends(lambda: get_redis_with_db(db=1))):
    # Usa DB 1 para caching
    pass


@app.get("/session-data")
async def session_endpoint(r: Redis = Depends(lambda: get_redis_with_db(db=2))):
    # Usa DB 2 para sessions
    pass

3. Combinación con otras dependencies:

async def get_current_user(token: str = Depends(oauth2_scheme), r: Redis = Depends(get_redis)):
    """Validar JWT + verificar session en Redis (juntos)."""
    payload = decode_jwt(token)
    session = await r.hgetall(f"session:{payload['session_id']}")
    if not session:
        raise HTTPException(401)
    return {"user_id": int(session["user_id"]), "session_id": payload["session_id"]}


@app.get("/me")
async def me(user: dict = Depends(get_current_user)):
    return user

El patrón en redis_client.py

"""
Singleton + dependency en un solo módulo.
"""
from redis.asyncio import Redis, ConnectionPool


_pool: ConnectionPool | None = None
_client: Redis | None = None


def init_pool(url: str = "redis://localhost:6379/0", max_connections: int = 50):
    global _pool, _client
    _pool = ConnectionPool.from_url(
        url,
        max_connections=max_connections,
        decode_responses=True,
    )
    _client = Redis(connection_pool=_pool)


async def close_pool():
    global _pool, _client
    if _client:
        await _client.aclose()
    if _pool:
        await _pool.aclose()
    _pool, _client = None, None


# La dependency: simple
async def get_redis() -> Redis:
    if _client is None:
        raise RuntimeError("Redis pool not initialized. Did you forget to call init_pool() in lifespan?")
    return _client

Patrón 2: Lifespan Events

Inicialización moderna con lifespan

FastAPI deprecated @app.on_event("startup") en favor de lifespan context manager (más limpio y testeable).

from contextlib import asynccontextmanager
from fastapi import FastAPI

from redis_client import init_pool, close_pool, get_redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # === Startup ===
    init_pool(url="redis://localhost:6379/0", max_connections=100)

    # Verificar conexión al startup (fail fast)
    r = await get_redis()
    try:
        await r.ping()
        print("✓ Redis connected")
    except Exception as e:
        print(f"✗ Redis connection failed: {e}")
        raise

    yield  # ← App ejecuta aquí

    # === Shutdown ===
    print("Shutting down: closing Redis pool...")
    await close_pool()


app = FastAPI(
    title="My API",
    lifespan=lifespan,
)

Ventajas

1. Fail fast: si Redis no está disponible al startup, la app no arranca (mejor que crash en runtime)
2. Cleanup limpio: el pool se cierra cuando uvicorn recibe SIGTERM
3. Testeable: las funciones startup/shutdown son explícitas

Lifespan con múltiples resources

Si tu app tiene Redis + DB + otros recursos:

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Redis
    init_redis_pool()
    print("✓ Redis pool initialized")

    # DB
    init_db_pool()
    print("✓ DB pool initialized")

    # Background tasks
    background_task = asyncio.create_task(periodic_cleanup())
    print("✓ Background tasks started")

    yield

    # Shutdown en orden inverso
    background_task.cancel()
    await close_db_pool()
    await close_redis_pool()
    print("✓ All resources closed")

Patrón 3: Middleware de Caching Automático

Hasta ahora, cada endpoint que quería cache tenía que escribir el patrón completo: r.get → if None → query DB → r.set. Un middleware de caching abstrae esto: cualquier endpoint GET puede ser cacheado sin que el código del endpoint cambie.

Implementación básica

"""
Middleware: cachea respuestas de GET endpoints automáticamente.
"""
import hashlib
import json
import logging
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response, JSONResponse
from redis.exceptions import RedisError


logger = logging.getLogger(__name__)


class CacheMiddleware(BaseHTTPMiddleware):
    """
    Cachea responses de endpoints GET con TTL configurable.

    Configuración:
    - cacheable_paths: prefijos de paths a cachear (ej. ["/api/products", "/api/categories"])
    - default_ttl: TTL default si el endpoint no especifica Cache-Control
    """

    def __init__(self, app, redis_client_getter, cacheable_paths: list, default_ttl: int = 300):
        super().__init__(app)
        self.get_redis = redis_client_getter
        self.cacheable_paths = cacheable_paths
        self.default_ttl = default_ttl

    async def dispatch(self, request: Request, call_next):
        # Solo cachear GETs
        if request.method != "GET":
            return await call_next(request)

        # Solo paths cacheables
        if not any(request.url.path.startswith(p) for p in self.cacheable_paths):
            return await call_next(request)

        # Generar cache key
        cache_key = self._build_cache_key(request)

        # Try cache
        try:
            r = self.get_redis()
            cached = await r.get(cache_key)
            if cached:
                logger.debug(f"Cache HIT: {cache_key}")
                return Response(
                    content=cached,
                    media_type="application/json",
                    headers={"X-Cache": "HIT"},
                )
        except RedisError as e:
            logger.warning(f"Cache read failed for {cache_key}: {e}")

        # Cache miss: ejecutar el endpoint
        response = await call_next(request)

        # Cachear el response (solo si fue exitoso)
        if response.status_code == 200:
            try:
                body = b""
                async for chunk in response.body_iterator:
                    body += chunk

                # Determinar TTL del response
                cache_control = response.headers.get("cache-control", "")
                ttl = self._extract_ttl(cache_control) or self.default_ttl

                r = self.get_redis()
                await r.set(cache_key, body, ex=ttl)

                # Re-construir el response (porque ya consumimos el body_iterator)
                return Response(
                    content=body,
                    status_code=response.status_code,
                    headers={**dict(response.headers), "X-Cache": "MISS"},
                    media_type=response.media_type,
                )
            except RedisError as e:
                logger.warning(f"Cache write failed for {cache_key}: {e}")
                return response

        return response

    def _build_cache_key(self, request: Request) -> str:
        """Construye una cache key única para la request."""
        # Incluye path y query params (orderless)
        params = sorted(request.query_params.items())
        params_str = "&".join(f"{k}={v}" for k, v in params)
        raw_key = f"{request.url.path}?{params_str}"

        # Hash para keys cortas
        h = hashlib.sha256(raw_key.encode()).hexdigest()[:16]
        return f"cache:http:{request.url.path}:{h}"

    def _extract_ttl(self, cache_control: str) -> int | None:
        """Extrae max-age de un Cache-Control header. None si no hay."""
        if "max-age=" in cache_control:
            try:
                ttl = int(cache_control.split("max-age=")[1].split(",")[0].strip())
                return ttl
            except (IndexError, ValueError):
                pass
        return None

Uso en la app

from redis_client import get_redis_client


app.add_middleware(
    CacheMiddleware,
    redis_client_getter=get_redis_client,  # devuelve el Redis singleton (no es coroutine)
    cacheable_paths=["/api/products", "/api/categories"],
    default_ttl=300,  # 5 min
)

(El redis_client_getter es un callable sync que devuelve el cliente Redis ya inicializado. Como el middleware lo invoca en cada request, evita la sobrecarga de DI por request.)

Header Cache-Control controla el TTL desde endpoint

@app.get("/api/products")
async def list_products(response: Response):
    response.headers["Cache-Control"] = "max-age=600"  # 10 min para esta key
    return {"products": [...]}


@app.get("/api/categories")
async def list_categories(response: Response):
    response.headers["Cache-Control"] = "max-age=3600"  # 1 hora
    return {"categories": [...]}

El middleware lee el header y respeta el TTL específico del endpoint.

Skip cache con header

# Cliente puede forzar cache miss con header
GET /api/products
Cache-Control: no-cache

# El middleware respeta esto

async def dispatch(self, request, call_next):
    # ... existing code ...

    # Skip cache si el cliente lo pide
    if "no-cache" in request.headers.get("cache-control", ""):
        return await call_next(request)

Verificación con curl

# First request: cache MISS
curl -i http://localhost:8000/api/products
# X-Cache: MISS

# Second request: cache HIT
curl -i http://localhost:8000/api/products
# X-Cache: HIT

# Force MISS
curl -i -H "Cache-Control: no-cache" http://localhost:8000/api/products
# X-Cache: MISS (regenerated)

Patrón 4: Graceful Degradation Completa

La app debe seguir funcionando cuando Redis falla. Lo viste con cache-aside; ahora lo aplicas a todo: rate limiting, sessions, caching middleware, todo.

Estrategia clara por componente

Ejemplo: rate limiting con degradation

async def check_rate_limit_safe(user_id: str, ...) -> tuple[bool, dict]:
    try:
        r = get_redis_client()
        # ... sliding window logic ...
        return (allowed, info)
    except RedisError as e:
        logger.warning(f"Rate limiter unavailable: {e}. Allowing request.")
        return (True, {"degraded": True})

Endpoint health diferenciado

@app.get("/health")
async def health():
    redis_status = await check_redis_health()

    # Si Redis está down pero la app puede degradar, retornamos 200 con status degraded
    if redis_status["status"] != "healthy":
        return JSONResponse(
            status_code=200,  # o 503 dependiendo de tu política
            content={
                "status": "degraded",
                "services": {"redis": redis_status},
                "message": "API operational but cache/sessions unavailable",
            }
        )

    return {"status": "healthy", "services": {"redis": redis_status}}

Política recomendada:

• 200 con "degraded": la app sigue respondiendo (lo prefiero)
• 503 Service Unavailable: orquestador puede reiniciar el contenedor

Decide según tu setup. Si Redis cae frecuentemente, "degraded" es más resiliente. Si Redis caer es indicador de un problema mayor, 503 fuerza al orquestador a actuar.

App completa con todos los patterns

Vamos a ensamblar todo. Crea app/main.py:

"""
FastAPI app con todos los patterns del módulo 4.
"""
import json
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, Request, Response
from redis.asyncio import Redis
from redis.exceptions import RedisError

from app.redis_client import init_pool, close_pool, get_redis_client
from app.cache_middleware import CacheMiddleware


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════════════════════
# Lifespan
# ═══════════════════════════════════════════════════════════


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    init_pool(url="redis://localhost:6379/0", max_connections=50)

    r = get_redis_client()
    try:
        await r.ping()
        logger.info("✓ Redis connected")
    except Exception as e:
        logger.error(f"✗ Redis connection failed: {e}")
        raise  # fail fast

    yield

    # Shutdown
    logger.info("Shutting down...")
    await close_pool()


# ═══════════════════════════════════════════════════════════
# App
# ═══════════════════════════════════════════════════════════


app = FastAPI(
    title="API with Redis Integration",
    lifespan=lifespan,
)


# Middleware de caching
app.add_middleware(
    CacheMiddleware,
    redis_client_getter=get_redis_client,
    cacheable_paths=["/api/products", "/api/categories"],
    default_ttl=300,
)


# ═══════════════════════════════════════════════════════════
# Dependencies
# ═══════════════════════════════════════════════════════════


async def get_redis() -> Redis:
    return get_redis_client()


# ═══════════════════════════════════════════════════════════
# Endpoints
# ═══════════════════════════════════════════════════════════


@app.get("/health")
async def health(r: Redis = Depends(get_redis)):
    try:
        await r.ping()
        info = await r.info("server")
        return {
            "status": "healthy",
            "services": {
                "redis": {
                    "status": "healthy",
                    "version": info.get("redis_version"),
                }
            }
        }
    except Exception as e:
        return {
            "status": "degraded",
            "services": {
                "redis": {"status": "unhealthy", "reason": str(e)}
            },
            "message": "API operational but cache unavailable",
        }


@app.get("/api/products")
async def list_products(response: Response, r: Redis = Depends(get_redis)):
    response.headers["Cache-Control"] = "max-age=600"

    # Mock DB query
    return {
        "products": [
            {"id": 1, "name": "Laptop"},
            {"id": 2, "name": "Phone"},
            {"id": 3, "name": "Tablet"},
        ]
    }


@app.get("/api/categories")
async def list_categories(response: Response, r: Redis = Depends(get_redis)):
    response.headers["Cache-Control"] = "max-age=3600"

    return {
        "categories": ["electronics", "books", "clothing"]
    }


@app.get("/api/products/{id}")
async def get_product(id: int, response: Response, r: Redis = Depends(get_redis)):
    """
    Cache-aside con DI.
    El middleware NO cachea esto (no en cacheable_paths) porque es dynamic.
    Aquí cacheamos manualmente.
    """
    cache_key = f"product:{id}"

    # Try cache
    try:
        cached = await r.get(cache_key)
        if cached:
            response.headers["X-Cache"] = "HIT"
            return json.loads(cached)
    except RedisError as e:
        logger.warning(f"Cache read failed: {e}")

    # Mock DB query
    product = {"id": id, "name": f"Product {id}", "price": 100 + id}

    # Try cache
    try:
        await r.set(cache_key, json.dumps(product), ex=300)
        response.headers["X-Cache"] = "MISS"
    except RedisError as e:
        logger.warning(f"Cache write failed: {e}")
        response.headers["X-Cache"] = "BYPASS"

    return product


@app.post("/api/products/{id}/invalidate")
async def invalidate_product(id: int, r: Redis = Depends(get_redis)):
    """Endpoint admin para invalidar cache de un producto + publish event."""
    try:
        # Invalidar cache local
        await r.delete(f"product:{id}")

        # Invalidar cache del middleware (por path)
        # Nota: en producción usarías un patrón más sofisticado
        cursor = 0
        while True:
            cursor, keys = await r.scan(cursor=cursor, match="cache:http:/api/products*", count=100)
            if keys:
                await r.delete(*keys)
            if cursor == 0:
                break

        # Publish invalidation event
        await r.publish(f"cache:invalidate:product:{id}", json.dumps({"product_id": id}))

        return {"invalidated": id}
    except RedisError as e:
        logger.error(f"Invalidation failed: {e}")
        return {"error": "invalidation failed", "fallback": "cache will expire naturally"}

Verificación

# Setup
cd ~/projects/redis-guide/module-04-async-pubsub
mkdir -p app && cd app
# (copiar redis_client.py, cache_middleware.py, main.py)

uvicorn app.main:app --reload &
sleep 2

# Health
curl -s http://localhost:8000/health | jq

# First request: cache MISS
curl -i http://localhost:8000/api/products | grep X-Cache
# X-Cache: MISS

# Second request: cache HIT (middleware lo sirvió)
curl -i http://localhost:8000/api/products | grep X-Cache
# X-Cache: HIT

# Different endpoint
curl -i http://localhost:8000/api/products/42 | grep X-Cache

# Detener Redis
docker stop redis-dev

# La app sigue funcionando (degraded)
curl -s http://localhost:8000/api/products | jq
# {"products": [...]}  ← funciona (sin cache)

curl -s http://localhost:8000/health | jq
# {"status": "degraded", ...}

docker start redis-dev

Troubleshooting

Problema 1: RuntimeError: Redis pool not initialized

Causa: Llamaste get_redis() antes del lifespan startup.

Solución: Asegúrate de que init_pool() está en el lifespan, y de que estás usando app = FastAPI(lifespan=lifespan).

Problema 2: Middleware cachea responses con errores 500

Causa: El middleware no verifica status_code.

Solución: Verifica response.status_code == 200 antes de cachear:

if response.status_code == 200:
    # cachear

Problema 3: Cache key includes auth token (cache leak)

Causa: Usas headers como Authorization en la cache key. Cada usuario tiene su propio cache, ineficiente.

Solución: Solo path + query params en la cache key. Si el endpoint depende de auth (dato del usuario), NO debería cachearse globalmente — es per-user, lo manejas en el endpoint.

Problema 4: Body iterator se consume y rompe el response

Causa: Después de leer response.body_iterator, el response no puede usarse.

Solución: Re-construir el Response después de consumir el body (como en el código del middleware).

Problema 5: Lifespan not called in tests

Causa: TestClient no ejecuta lifespan por default.

Solución:

from fastapi.testclient import TestClient

# Modern: lifespan se ejecuta automáticamente
client = TestClient(app)

# Para tests async:
from httpx import AsyncClient

async with AsyncClient(app=app, base_url="http://test") as ac:
    response = await ac.get("/health")

Si tu lifespan tiene side effects que no quieres en tests, mock las funciones:

@pytest.fixture
def app_no_redis():
    # Override init_pool to no-op
    ...

Problema 6: Multiple workers de uvicorn no comparten cache

Causa: Si tienes 4 workers con --workers 4, cada uno tiene su propio singleton Python — pero TODOS apuntan al mismo Redis.

Solución: Esto es CORRECTO. El singleton es por worker, pero todos los workers comparten Redis. La cache funciona porque las keys están en Redis (compartido), no en memoria del proceso. Todo bien.

Ejercicios

Ejercicio 1: Lifespan + Dependency injection (Fácil)

Crea una app FastAPI con lifespan que inicializa el pool, una dependency get_redis(), y un endpoint /health que use la dependency.

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from redis.asyncio import Redis, ConnectionPool


_pool = None
_client = None


@asynccontextmanager
async def lifespan(app):
    global _pool, _client
    _pool = ConnectionPool.from_url("redis://localhost:6379/0", max_connections=20, decode_responses=True)
    _client = Redis(connection_pool=_pool)
    await _client.ping()
    yield
    await _client.aclose()
    await _pool.aclose()


async def get_redis() -> Redis:
    return _client


app = FastAPI(lifespan=lifespan)


@app.get("/health")
async def health(r: Redis = Depends(get_redis)):
    try:
        await r.ping()
        return {"status": "healthy"}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}

uvicorn app:app --reload
curl http://localhost:8000/health
# {"status":"healthy"}

Ejercicio 2: Cache middleware básico (Medio)

Implementa un middleware simple que cachee responses de paths específicos por 60 segundos. Test que la segunda request retorna X-Cache: HIT.

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import Response


class SimpleCacheMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, redis_getter, paths: list, ttl: int = 60):
        super().__init__(app)
        self.get_redis = redis_getter
        self.paths = paths
        self.ttl = ttl

    async def dispatch(self, request, call_next):
        if request.method != "GET":
            return await call_next(request)

        if not any(request.url.path.startswith(p) for p in self.paths):
            return await call_next(request)

        key = f"cache:{request.url.path}"
        r = self.get_redis()

        try:
            cached = await r.get(key)
            if cached:
                return Response(content=cached, media_type="application/json", headers={"X-Cache": "HIT"})
        except Exception:
            pass

        response = await call_next(request)

        if response.status_code == 200:
            body = b""
            async for chunk in response.body_iterator:
                body += chunk
            try:
                await r.set(key, body, ex=self.ttl)
            except Exception:
                pass
            return Response(content=body, status_code=200, media_type=response.media_type, headers={**dict(response.headers), "X-Cache": "MISS"})

        return response

app.add_middleware(SimpleCacheMiddleware, redis_getter=lambda: _client, paths=["/api/products"], ttl=60)


@app.get("/api/products")
async def products():
    return {"products": ["a", "b"]}

curl -i http://localhost:8000/api/products | grep X-Cache
# X-Cache: MISS
curl -i http://localhost:8000/api/products | grep X-Cache
# X-Cache: HIT

Ejercicio 3: Health check con Redis verificación (Medio)

Implementa /health que: si Redis está OK retorna 200 con info detallada, si Redis falla retorna 200 con status "degraded" y razón.

import asyncio
from fastapi import Depends
from redis.exceptions import RedisError


@app.get("/health")
async def health(r: Redis = Depends(get_redis)):
    health_data = {
        "status": "healthy",
        "services": {},
    }

    # Redis check
    try:
        await asyncio.wait_for(r.ping(), timeout=1.0)
        info = await r.info("server")
        memory = await r.info("memory")
        clients = await r.info("clients")

        health_data["services"]["redis"] = {
            "status": "healthy",
            "version": info.get("redis_version"),
            "uptime_seconds": int(info.get("uptime_in_seconds", 0)),
            "memory_used": memory.get("used_memory_human"),
            "connected_clients": clients.get("connected_clients"),
        }
    except (RedisError, asyncio.TimeoutError) as e:
        health_data["status"] = "degraded"
        health_data["services"]["redis"] = {
            "status": "unhealthy",
            "reason": str(e),
        }
        health_data["message"] = "Redis unavailable. App operating without cache."

    return health_data

curl -s http://localhost:8000/health | jq
# {
#   "status": "healthy",
#   "services": {
#     "redis": {
#       "status": "healthy",
#       "version": "7.2.4",
#       "uptime_seconds": 12345,
#       "memory_used": "1.5M",
#       "connected_clients": 1
#     }
#   }
# }

docker stop redis-dev

curl -s http://localhost:8000/health | jq
# {
#   "status": "degraded",
#   "services": {
#     "redis": {"status": "unhealthy", ...}
#   },
#   "message": "Redis unavailable..."
# }

Ejercicio 4: Endpoint con Cache-Control desde el endpoint (Medio)

Implementa un endpoint que setee Cache-Control: max-age=120 desde el handler. Verifica con tu middleware que respeta el TTL del header (no usa el default).

@app.get("/api/short-cached")
async def short_cached(response: Response):
    response.headers["Cache-Control"] = "max-age=10"   # 10 seconds
    return {"data": "this caches for 10s"}


@app.get("/api/long-cached")
async def long_cached(response: Response):
    response.headers["Cache-Control"] = "max-age=3600"  # 1 hour
    return {"data": "this caches for 1h"}


# Asume que el middleware ahora respeta Cache-Control header

# Short
curl -i http://localhost:8000/api/short-cached
# Cache-Control: max-age=10
# X-Cache: MISS (first)

curl -i http://localhost:8000/api/short-cached
# X-Cache: HIT

# Verificar TTL en Redis
redis-cli TTL "cache:http:/api/short-cached:abc123..."
# (integer) 9   ← cerca de 10

Ejercicio 5: Graceful degradation completa (Difícil)

Implementa una API mock con 3 endpoints (/products, /users, /orders) que use Redis para caching. Verifica que cuando Redis está caído los 3 endpoints siguen retornando data correctamente (con warnings en logs).

import json
import logging
from fastapi import FastAPI, Depends, Response
from redis.asyncio import Redis
from redis.exceptions import RedisError


logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


fake_db = {
    "products": [{"id": i, "name": f"Product {i}"} for i in range(1, 6)],
    "users": [{"id": i, "name": f"User {i}"} for i in range(1, 6)],
    "orders": [{"id": i, "total": i * 100} for i in range(1, 6)],
}


async def cached_or_fallback(r: Redis, key: str, fallback, ttl: int = 60):
    # Try read
    try:
        cached = await r.get(key)
        if cached:
            return json.loads(cached), "HIT"
    except RedisError as e:
        logger.warning(f"Cache read failed for {key}: {e}")

    # Compute
    data = fallback()

    # Try write
    try:
        await r.set(key, json.dumps(data), ex=ttl)
    except RedisError as e:
        logger.warning(f"Cache write failed for {key}: {e}")

    return data, "MISS"


@app.get("/products")
async def list_products(response: Response, r: Redis = Depends(get_redis)):
    data, status = await cached_or_fallback(r, "cache:products", lambda: fake_db["products"])
    response.headers["X-Cache"] = status
    return data


@app.get("/users")
async def list_users(response: Response, r: Redis = Depends(get_redis)):
    data, status = await cached_or_fallback(r, "cache:users", lambda: fake_db["users"])
    response.headers["X-Cache"] = status
    return data


@app.get("/orders")
async def list_orders(response: Response, r: Redis = Depends(get_redis)):
    data, status = await cached_or_fallback(r, "cache:orders", lambda: fake_db["orders"])
    response.headers["X-Cache"] = status
    return data

# Redis up: cache works
curl -i http://localhost:8000/products | grep X-Cache  # MISS
curl -i http://localhost:8000/products | grep X-Cache  # HIT

# Stop Redis
docker stop redis-dev

# All endpoints still work
curl -s http://localhost:8000/products | jq
curl -s http://localhost:8000/users | jq
curl -s http://localhost:8000/orders | jq

# Logs show warnings:
# WARNING:Cache read failed for cache:products: Connection refused
# WARNING:Cache write failed for cache:products: Connection refused

docker start redis-dev

Ejercicio 6: Test concurrente del setup completo (Difícil)

Lanza 200 requests concurrentes a un endpoint cacheado mientras Redis está caído. Verifica que las 200 retornen 200 OK (con warnings).

# test_degradation.py
import asyncio
import httpx


async def make_request(client, i):
    try:
        r = await client.get("http://localhost:8000/products")
        return r.status_code
    except Exception as e:
        return f"error: {e}"


async def main():
    # Stop Redis primero
    print("Make sure Redis is STOPPED before running this test!")
    print("(docker stop redis-dev)")
    input("Press Enter when Redis is stopped...")

    async with httpx.AsyncClient() as client:
        tasks = [make_request(client, i) for i in range(200)]
        results = await asyncio.gather(*tasks)

    counts = {}
    for r in results:
        counts[r] = counts.get(r, 0) + 1

    print(f"\nResults: {counts}")

    if counts.get(200, 0) == 200:
        print("✅ Graceful degradation funcionando: 200/200 sin Redis")
    else:
        print("⚠️ Algunas requests fallaron")


asyncio.run(main())

Make sure Redis is STOPPED before running this test!
Press Enter when Redis is stopped...

Results: {200: 200}
✅ Graceful degradation funcionando: 200/200 sin Redis

Resumen

En esta cápsula aprendiste:

4 patterns profesionales para FastAPI + Redis:

1. Dependency Injection con Depends(get_redis):

   • Endpoints reciben Redis sin acoplarse a singleton global
   • Tests fáciles con app.dependency_overrides
   • Combinable con otras dependencies (auth, etc.)

2. Lifespan Events:

   • init_pool() en startup, close_pool() en shutdown
   • Fail fast: app no arranca si Redis no responde
   • Cleanup limpio en SIGTERM

3. Middleware de Caching:

   • Cachea responses GET automáticamente
   • Respeta Cache-Control: max-age= del endpoint
   • Header X-Cache: HIT/MISS para debugging
   • Skip con Cache-Control: no-cache

4. Graceful Degradation:

   • try/except en cada operación Redis
   • Cache falla → ir a DB directamente
   • Rate limit falla → permitir request (warning log)
   • Health check reporta degraded (no crash)

App completa:

• Lifespan inicializa pool con max_connections=50
• Endpoints usan Depends(get_redis)
• Middleware cachea /api/products y /api/categories
• Health check con verificación de Redis
• App funciona aunque Redis esté caído (degraded)

El stack que el módulo 5 asume:

• redis-py >= 5.0 con redis.asyncio
• ConnectionPool con singleton
• DI con Depends(get_redis)
• Middleware de caching automático
• Health check + graceful degradation

Recursos adicionales

1. FastAPI Lifespan Events — Pattern moderno para resources
2. FastAPI Dependencies — DI explicado en docs oficiales
3. Starlette Middleware — Base middleware que FastAPI usa
4. HTTP Cache-Control RFC — Spec del header Cache-Control
5. Twelve-Factor App: Disposability — Por qué cleanup limpio importa
6. redis-py Health Check Best Practices — Patterns para production health checks

¿Qué sigue?

En la Cápsula 05 consolidas todo el módulo 4 en el Mini-proyecto Real-time Notifications: una app que combina Pub/Sub + WebSocket + FastAPI integration. El publisher emite eventos cuando datos cambian; el subscriber los recibe y los broadcastea a clientes WebSocket conectados. Es un mini-proyecto portfolio-worthy y prepara directamente para el módulo 5 (Production Cached API).

Mantén Redis corriendo. Adelante.