Deep FastAPI Integration

Overview

This capsule consolidates everything you learned in module 4 into the professional integration pattern with FastAPI: the technical scaffolding every production app should have to use Redis correctly. You'll learn 4 patterns that separate toy code from portfolio-worthy code: dependency injection with Depends(get_redis) so every endpoint receives the client without coupling to a global singleton, lifespan events that initialize the pool at startup and close it cleanly at shutdown, automatic caching middleware that caches responses without each endpoint having to think about it, and full graceful degradation with structured logging.

This is module 4's capsule 04: the one that ties the threads together. Capsule 02 showed Pub/Sub. Capsule 03 showed redis.asyncio and connection pooling. Here you integrate them inside FastAPI with the framework's idiomatic patterns. By the end of the capsule, you'll have a FastAPI app with Redis professionally integrated: a pool initialized at startup, dependency injection in every endpoint, middleware that caches GET endpoints automatically, a health check that reports Redis's status, and graceful degradation when Redis fails. It's the scaffolding module 5 (Production Cached API) assumes from day 1.

There are no heavy new concepts here — they're combinations of patterns you've already seen, assembled professionally. But the difference between knowing the individual concepts and knowing how to integrate them cohesively is what separates a junior developer from a senior one. This capsule is where you make that jump.

The final stack

The FastAPI app
    │
    ├── Lifespan events: init_pool() on startup, close_pool() on shutdown
    │
    ├── Dependency injection: Depends(get_redis) in the endpoints
    │
    ├── Middleware: automatic caching for GET endpoints
    │
    ├── Routes:
    │   ├── /health (checks that Redis is up)
    │   ├── /api/* (with automatic caching via the middleware)
    │   └── /admin/* (admin operations)
    │
    └── Graceful degradation: try/except around every Redis operation

Pattern 1: Dependency Injection

Why DI instead of a direct import

# ❌ Anti-pattern: importing the singleton directly
from redis_client import get_redis


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

It works, but it has problems:

1. Hard to test: to mock Redis, you have to mock the whole module
2. Coupling: every endpoint knows the singleton's implementation
3. It isn't idiomatic in FastAPI: the framework provides DI for a reason

The Depends pattern

from fastapi import Depends
from redis.asyncio import Redis

from redis_client import get_redis_client


# The dependency: a function that returns the client
async def get_redis() -> Redis:
    return get_redis_client()


# The endpoint uses 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)
    # ...

The real benefits

1. Tests with clean mocks:

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. Injection with per-endpoint configuration:

async def get_redis_with_db(db: int = 0):
    """A different Redis DB for different 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))):
    # Uses DB 1 for caching
    pass


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

3. Combining it with other dependencies:

async def get_current_user(token: str = Depends(oauth2_scheme), r: Redis = Depends(get_redis)):
    """Validate the JWT + verify the session in Redis (together)."""
    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

The pattern in redis_client.py

"""
The singleton + the dependency in a single module.
"""
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


# The 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

Pattern 2: Lifespan Events

Modern initialization with lifespan

FastAPI deprecated @app.on_event("startup") in favor of the lifespan context manager (cleaner and more testable).

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)

    # Verify the connection at 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  # ← The app runs here

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


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

The advantages

1. Fail fast: if Redis isn't available at startup, the app doesn't start (better than crashing at runtime)
2. Clean cleanup: the pool closes when uvicorn receives SIGTERM
3. Testable: the startup/shutdown functions are explicit

Lifespan with multiple resources

If your app has Redis + a DB + other resources:

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

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

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

    yield

    # Shutdown in reverse order
    background_task.cancel()
    await close_db_pool()
    await close_redis_pool()
    print("✓ All resources closed")

Pattern 3: Automatic Caching Middleware

Until now, every endpoint that wanted caching had to write the full pattern: r.get → if None → query the DB → r.set. A caching middleware abstracts this away: any GET endpoint can be cached without the endpoint's code changing.

A basic implementation

"""
Middleware: caches GET endpoint responses automatically.
"""
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):
    """
    Caches GET endpoint responses with a configurable TTL.

    Configuration:
    - cacheable_paths: path prefixes to cache (e.g., ["/api/products", "/api/categories"])
    - default_ttl: the default TTL if the endpoint doesn't specify 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):
        # Only cache GETs
        if request.method != "GET":
            return await call_next(request)

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

        # Generate the cache key
        cache_key = self._build_cache_key(request)

        # Try the 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: run the endpoint
        response = await call_next(request)

        # Cache the response (only if it succeeded)
        if response.status_code == 200:
            try:
                body = b""
                async for chunk in response.body_iterator:
                    body += chunk

                # Determine the response's TTL
                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)

                # Rebuild the response (because we already consumed the 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:
        """Builds a unique cache key for the request."""
        # It includes the path and the query params (order-independent)
        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}"

        # A hash for short keys
        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:
        """Extracts max-age from a Cache-Control header. None if there isn't one."""
        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

Using it in the app

from redis_client import get_redis_client


app.add_middleware(
    CacheMiddleware,
    redis_client_getter=get_redis_client,  # returns the Redis singleton (it isn't a coroutine)
    cacheable_paths=["/api/products", "/api/categories"],
    default_ttl=300,  # 5 min
)

(The redis_client_getter is a sync callable that returns the already-initialized Redis client. Since the middleware invokes it on every request, this avoids per-request DI overhead.)

The Cache-Control header controls the TTL from the endpoint

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


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

The middleware reads the header and honors the endpoint's specific TTL.

Skipping the cache with a header

# The client can force a cache miss with a header
GET /api/products
Cache-Control: no-cache

# The middleware honors this

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

    # Skip the cache if the client asks for it
    if "no-cache" in request.headers.get("cache-control", ""):
        return await call_next(request)

Verifying with 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 a MISS
curl -i -H "Cache-Control: no-cache" http://localhost:8000/api/products
# X-Cache: MISS (regenerated)

Pattern 4: Full Graceful Degradation

The app should keep working when Redis fails. You saw it with cache-aside; now you apply it to everything: rate limiting, sessions, the caching middleware, all of it.

A clear strategy per component

An example: rate limiting with 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})

A differentiated health endpoint

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

    # If Redis is down but the app can degrade, we return a 200 with a degraded status
    if redis_status["status"] != "healthy":
        return JSONResponse(
            status_code=200,  # or 503 depending on your policy
            content={
                "status": "degraded",
                "services": {"redis": redis_status},
                "message": "API operational but cache/sessions unavailable",
            }
        )

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

The recommended policy:

• 200 with "degraded": the app keeps responding (my preference)
• 503 Service Unavailable: the orchestrator can restart the container

Decide based on your setup. If Redis goes down frequently, "degraded" is more resilient. If Redis going down is a sign of a bigger problem, a 503 forces the orchestrator to act.

The complete app with all the patterns

Let's assemble everything. Create app/main.py:

"""
A FastAPI app with all of module 4's patterns.
"""
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,
)


# The caching middleware
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"

    # A 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 with DI.
    The middleware does NOT cache this (it isn't in cacheable_paths) because it's dynamic.
    Here we cache it manually.
    """
    cache_key = f"product:{id}"

    # Try the 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}")

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

    # Try to cache it
    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)):
    """An admin endpoint to invalidate a product's cache + publish an event."""
    try:
        # Invalidate the local cache
        await r.delete(f"product:{id}")

        # Invalidate the middleware's cache (by path)
        # Note: in production you'd use a more sophisticated pattern
        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 the 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"}

Verification

# Setup
cd ~/projects/redis-guide/module-04-async-pubsub
mkdir -p app && cd app
# (copy 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 (the middleware served it)
curl -i http://localhost:8000/api/products | grep X-Cache
# X-Cache: HIT

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

# Stop Redis
docker stop redis-dev

# The app keeps working (degraded)
curl -s http://localhost:8000/api/products | jq
# {"products": [...]}  ← it works (without cache)

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

docker start redis-dev

Troubleshooting

Problem 1: RuntimeError: Redis pool not initialized

Cause: You called get_redis() before the lifespan startup.

Solution: Make sure init_pool() is in the lifespan, and that you're using app = FastAPI(lifespan=lifespan).

Problem 2: The middleware caches responses with 500 errors

Cause: The middleware doesn't check the status_code.

Solution: Check response.status_code == 200 before caching:

if response.status_code == 200:
    # cache it

Problem 3: The cache key includes the auth token (a cache leak)

Cause: You use headers like Authorization in the cache key. Every user gets their own cache — inefficient.

Solution: Only the path + query params in the cache key. If the endpoint depends on auth (user-specific data), it should NOT be cached globally — it's per-user, and you handle it in the endpoint.

Problem 4: The body iterator gets consumed and breaks the response

Cause: After reading response.body_iterator, the response can't be used.

Solution: Rebuild the Response after consuming the body (like in the middleware's code).

Problem 5: The lifespan isn't called in tests

Cause: TestClient doesn't run the lifespan by default.

Solution:

from fastapi.testclient import TestClient

# Modern: the lifespan runs automatically
client = TestClient(app)

# For async tests:
from httpx import AsyncClient

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

If your lifespan has side effects you don't want in tests, mock the functions:

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

Problem 6: Multiple uvicorn workers don't share the cache

Cause: If you have 4 workers with --workers 4, each has its own Python singleton — but they ALL point to the same Redis.

Solution: This is CORRECT. The singleton is per worker, but every worker shares Redis. The cache works because the keys are in Redis (shared), not in the process's memory. All good.

Exercises

Exercise 1: Lifespan + dependency injection (Easy)

Create a FastAPI app with a lifespan that initializes the pool, a get_redis() dependency, and a /health endpoint that uses the 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"}

Exercise 2: A basic cache middleware (Medium)

Implement a simple middleware that caches responses from specific paths for 60 seconds. Test that the second request returns 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

Exercise 3: A health check with Redis verification (Medium)

Implement /health so that: if Redis is OK it returns a 200 with detailed info, and if Redis fails it returns a 200 with the status "degraded" and the reason.

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": {},
    }

    # The 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..."
# }

Exercise 4: An endpoint with Cache-Control set from the endpoint (Medium)

Implement an endpoint that sets Cache-Control: max-age=120 from the handler. Verify with your middleware that it honors the header's TTL (it doesn't use the 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"}


# This assumes the middleware now honors the 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

# Check the TTL in Redis
redis-cli TTL "cache:http:/api/short-cached:abc123..."
# (integer) 9   ← close to 10

Exercise 5: Full graceful degradation (Hard)

Implement a mock API with 3 endpoints (/products, /users, /orders) that uses Redis for caching. Verify that when Redis is down all 3 endpoints keep returning data correctly (with warnings in the 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: the 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 the endpoints still work
curl -s http://localhost:8000/products | jq
curl -s http://localhost:8000/users | jq
curl -s http://localhost:8000/orders | jq

# The 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

Exercise 6: A concurrent test of the complete setup (Hard)

Fire 200 concurrent requests at a cached endpoint while Redis is down. Verify all 200 return a 200 OK (with 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 first
    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 working: 200/200 without Redis")
    else:
        print("⚠️ Some requests failed")


asyncio.run(main())

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

Results: {200: 200}
✅ Graceful degradation working: 200/200 without Redis

Summary

In this capsule you learned:

4 professional patterns for FastAPI + Redis:

1. Dependency Injection with Depends(get_redis):

   • Endpoints receive Redis without coupling to a global singleton
   • Easy tests with app.dependency_overrides
   • Combinable with other dependencies (auth, etc.)

2. Lifespan Events:

   • init_pool() on startup, close_pool() on shutdown
   • Fail fast: the app doesn't start if Redis doesn't respond
   • Clean cleanup on SIGTERM

3. Caching Middleware:

   • It caches GET responses automatically
   • It honors the endpoint's Cache-Control: max-age=
   • An X-Cache: HIT/MISS header for debugging
   • Skippable with Cache-Control: no-cache

4. Graceful Degradation:

   • try/except around every Redis operation
   • The cache fails → go to the DB directly
   • Rate limiting fails → allow the request (a warning log)
   • The health check reports degraded (no crash)

The complete app:

• The lifespan initializes the pool with max_connections=50
• The endpoints use Depends(get_redis)
• The middleware caches /api/products and /api/categories
• A health check with a Redis check
• The app works even with Redis down (degraded)

The stack module 5 assumes:

• redis-py >= 5.0 with redis.asyncio
• A ConnectionPool with a singleton
• DI with Depends(get_redis)
• Automatic caching middleware
• A health check + graceful degradation

Additional resources

1. FastAPI Lifespan Events — The modern pattern for resources
2. FastAPI Dependencies — DI explained in the official docs
3. Starlette Middleware — The base middleware FastAPI uses
4. HTTP Cache-Control RFC — The Cache-Control header spec
5. Twelve-Factor App: Disposability — Why clean cleanup matters
6. redis-py Health Check Best Practices — Patterns for production health checks

What's next?

In Capsule 05 you consolidate all of module 4 in the Real-time Notifications mini-project: an app that combines Pub/Sub + WebSockets + the FastAPI integration. The publisher emits events when data changes; the subscriber receives them and broadcasts them to connected WebSocket clients. It's a portfolio-worthy mini-project and it prepares you directly for module 5 (Production Cached API).

Keep Redis running. Let's go.