Managing Redis Connections with the FastAPI Lifespan

Your FastAPI service opens a new Redis connection on every request — or worse, imports a module-level client that was created before the event loop existed — and under load it either leaks sockets or raises got Future attached to a different loop. The fix is to give the application one redis.asyncio connection pool whose lifetime is bound to the ASGI process: built once during startup, shared across every request through dependency injection, and drained cleanly on shutdown. This walkthrough builds that ownership model with the modern lifespan context manager, injects a pooled client into path operations, and adds a readiness probe so orchestrators only route traffic once Redis is reachable.

# Prerequisites

• Redis 6.2+ or 7.x reachable from the app process.
• FastAPI 0.110+ and an ASGI server — uvicorn[standard] 0.29+.
• pip install fastapi "uvicorn[standard]" "redis>=5,<6" — the async client ships inside redis-py 5.x as redis.asyncio.
• Python 3.10+ with async/await; every Redis call in this pattern is awaited.
• A sense of your per-worker concurrency, which sets the pool's max_connections.

# Step-by-Step Implementation

# 1. Define an async lifespan context manager

The lifespan async context manager runs setup before the app accepts traffic and teardown after the last request drains. Everything before the yield is startup; everything after is shutdown. Registering it on the app replaces the deprecated @app.on_event hooks with a single scope where the pool lives.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # --- startup runs here (before yield) ---
    yield
    # --- shutdown runs here (after yield) ---

app = FastAPI(lifespan=lifespan)

# 2. Create the connection pool on startup

Build a single redis.asyncio.ConnectionPool inside the pool's own event loop — that is, inside lifespan, not at import time. Creating it here guarantees it binds to the running loop, which is exactly what avoids the cross-loop errors that module-level clients cause. The choice of async client over a blocking one is discussed in asyncio vs Sync redis-py Clients.

import redis.asyncio as redis

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = redis.ConnectionPool.from_url(
        "redis://127.0.0.1:6379/0",
        max_connections=100,        # ceiling for this worker process
        decode_responses=True,
        socket_connect_timeout=2,
        socket_timeout=2,
    )
    # ... store it (next step) ...
    yield

# 3. Store the pool on app.state

app.state is the canonical place to hang process-wide singletons. Storing the pool there — rather than a global variable — keeps it discoverable from request handlers and dependencies without import-time side effects. Create one Redis client bound to the pool so callers reuse connections instead of each opening their own.

    app.state.redis_pool = pool
    app.state.redis = redis.Redis(connection_pool=pool)
    yield

# 4. Inject the client with a dependency

Expose the shared client through a dependency that reads it off app.state via the request. Every path operation that declares this dependency borrows the same pooled client, so a burst of requests multiplexes over the pool instead of spawning connections. This clean per-request handoff pairs naturally with a cache-aside read in the handler.

from fastapi import Depends, Request

async def get_redis(request: Request) -> redis.Redis:
    # Yield the process-wide client; no per-request connection is created.
    return request.app.state.redis

@app.get("/products/{product_id}")
async def read_product(product_id: int, r: redis.Redis = Depends(get_redis)):
    cached = await r.get(f"product:{product_id}")
    if cached is None:
        cached = await load_and_cache(r, product_id)   # populate on miss
    return {"product": cached}

# 5. Close the pool gracefully on shutdown

After the yield, drain the pool with aclose() so in-flight connections are released and the process exits without leaking sockets or logging Task was destroyed but it is pending. Wrapping teardown in the same context manager guarantees it runs on every clean shutdown, including a SIGTERM from an orchestrator.

@asynccontextmanager
async def lifespan(app: FastAPI):
    pool = redis.ConnectionPool.from_url("redis://127.0.0.1:6379/0", max_connections=100)
    app.state.redis = redis.Redis(connection_pool=pool)
    try:
        yield
    finally:
        # Release every pooled connection before the loop stops.
        await app.state.redis.aclose()
        await pool.aclose()

# 6. Add a readiness health check

Expose a /healthz endpoint that issues PING so an orchestrator's readiness probe only sends traffic once Redis actually answers. A cheap round trip here prevents the app from serving requests during the window where the process is up but its dependency is not.

from fastapi import HTTPException

@app.get("/healthz")
async def healthz(r: redis.Redis = Depends(get_redis)):
    try:
        await r.ping()               # cheap round trip to prove reachability
    except redis.RedisError:
        raise HTTPException(status_code=503, detail="redis unreachable")
    return {"status": "ok"}

# Failure Modes

Client attached to a different event loop. A module-level redis.Redis(...) created at import time binds to whatever loop imported it; under uvicorn it then raises RuntimeError: ... attached to a different loop on the first await. Diagnose by grepping for a client constructed outside lifespan:

grep -rn "redis.Redis(\|from_url(" app/ | grep -v lifespan

Fix by moving all pool and client creation inside the lifespan function (Steps 2-3) so they bind to the loop uvicorn actually runs the app on.

Connections leak across reloads or never drain. The process count of Redis clients climbs after every deploy or --reload, eventually hitting maxclients. Diagnose against the live server:

redis-cli INFO clients | grep -E "connected_clients|maxclients"

Fix by ensuring the finally block calls await app.state.redis.aclose() and await pool.aclose() (Step 5); without it, a restarted worker orphans its old connections until they time out.

Traffic routed before Redis is reachable. Requests 503 for the first few seconds after a rollout because the app accepted traffic before its dependency answered. Diagnose by hitting the probe directly:

curl -f http://localhost:8000/healthz   # non-zero exit means not ready

Fix by wiring the orchestrator's readiness probe to /healthz (Step 6) so it withholds traffic until PING succeeds.

# Verification

Confirm the app opens exactly one pool per worker and reuses it. Start the server, fire concurrent requests, and watch that connection count stays bounded well under max_connections × workers:

uvicorn app.main:app --workers 4 &
seq 1 200 | xargs -P 50 -I{} curl -s http://localhost:8000/products/1 >/dev/null
redis-cli INFO clients | grep connected_clients   # bounded, not one-per-request

Confirm a clean shutdown releases connections rather than orphaning them:

redis-cli INFO clients | grep connected_clients   # note the count
kill -TERM %1                                      # graceful SIGTERM to uvicorn
redis-cli INFO clients | grep connected_clients   # should drop back down

# FAQ

# Should I use the lifespan, middleware, or a global client for Redis?

Use the lifespan. Middleware runs per request and would rebuild or re-fetch the client on every call, adding overhead and offering no place for one-time teardown. A module-level global is created at import time, before the event loop exists, which is the direct cause of cross-loop errors under uvicorn. The lifespan gives you exactly one construction and one destruction bound to the app's real event loop, which is precisely what a pooled resource needs.

# How large should max_connections be per worker?

Size it to the peak number of concurrent awaited Redis calls a single worker sustains, plus headroom — not your total request rate. Because the event loop multiplexes, most handlers hold a connection only for the microseconds of an awaited command, so even high request rates need far fewer connections than requests. Start around 100 per worker, load-test, and watch connected_clients; raise it only if you observe pool-wait latency, and remember the ceiling is per worker, so total connections are max_connections × workers.

# How does this interact with uvicorn workers and process forking?

Each uvicorn worker is a separate process with its own event loop, so each runs lifespan independently and builds its own pool — pools are never shared across the fork boundary. That is correct and desired: a connection pool is not fork-safe, so a child inheriting a parent's live sockets would corrupt them. Because every worker builds its own pool after forking, multiply max_connections by the worker count when sizing against the Redis server's maxclients.

# Why a PING health check instead of just checking the process is up?

A liveness signal that only proves the Python process is running will happily report healthy while Redis is unreachable, so the orchestrator routes traffic that immediately fails. A PING round trip proves the actual dependency answers, which is what readiness means. Keep it cheap and dependency-scoped: probe only Redis on /healthz, return 503 on any RedisError, and let the orchestrator withhold traffic until the round trip succeeds.

---

Up: Python Redis Integration Patterns

# Related

• Configuring the Django Redis Cache Backend
• Using Redis as a Celery Result Backend
• Python Redis Integration Patterns
• Tuning redis-py Connection Pools for High Concurrency