Redis with FastAPI

Add idiomatic Redis connection management, dependency-injection caching, rate limiting, and sessions to FastAPI apps.

FastAPI is a modern, high-performance web framework for building APIs with Python. The official fastapi-redis-sdk integrates Redis with FastAPI without any boilerplate. It manages connection pools through the application lifespan and exposes caching as injectable dependencies, including HTTP-native ETag, 304 Not Modified, and Cache-Control support. It also provides distributed rate limiting and server-side sessions.

The SDK is built on the redis-py client, so anything redis-py can do is still available to you alongside the caching helpers.

Requirements

See Requirements on the GitHub repo for the full set of dependencies used by fastapi-redis-sdk.

You also need a running Redis server. You can run one locally with Redis Open Source, use Docker, or connect to Redis Cloud.

Install and import

Use the following command to install fastapi-redis-sdk:

pip install fastapi-redis-sdk

Note that although you install the package as fastapi-redis-sdk, you import it as redis_fastapi:

from redis_fastapi import FastAPIRedis

Quick start

Attach Redis to your app with the fluent builder. The lifespan() call hooks into the FastAPI lifespan events to open a connection pool at startup and close it cleanly on shutdown. Inject AsyncRedisDep into your async endpoints to get a ready-to-use client:

from fastapi import FastAPI
from redis_fastapi import FastAPIRedis, AsyncRedisDep

app = FastAPI()
FastAPIRedis(app).lifespan()

@app.get("/items")
async def get_items(redis: AsyncRedisDep):
    return {"items": await redis.get("items")}

The builder wraps any existing lifespan, so multiple libraries can register their own startup and shutdown logic without conflicting.

Configuration

All settings are read from environment variables prefixed with REDIS_, or from a .env file in your project root. The simplest setup is a single connection URL:

export REDIS_URL=redis://user:pass@host:6379/0

Alternatively, you can configure individual fields:

export REDIS_HOST=redis.example.com
export REDIS_PORT=6380
export REDIS_PASSWORD=secret

When REDIS_URL is set, it takes precedence over the individual connection fields. The most commonly used variables are described in the table below:

Configuration is validated with Pydantic Settings, so invalid values (for example, a port outside 1–65535) fail fast at startup. For the full environment-variable reference, TLS options, and programmatic configuration, see the SDK configuration guide.

Caching

Enable caching by adding .caching() to the builder chain. The SDK then offers two complementary approaches that share the same connection pool:

Dependency-injection factories

cache(), cache_evict(), and cache_put() are dependency factories you attach to a route. Because all three share the same key_builder, a GET, DELETE, and PUT on the same path target the exact same cache key:

from fastapi import Depends, FastAPI
from redis_fastapi import FastAPIRedis, cache, cache_evict, cache_put, default_key_builder

app = FastAPI()
FastAPIRedis(app).lifespan().caching()

# READ - cache the GET response for 5 minutes
@app.get("/products/{product_id}", dependencies=[Depends(cache(ttl=300, eviction_group="products"))])
async def get_product(product_id: int):
    return await db.get_product(product_id)

# INVALIDATE - evict the cached entry when the product is deleted
@app.delete(
    "/products/{product_id}",
    dependencies=[Depends(cache_evict(eviction_group="products", key_builder=default_key_builder))],
)
async def delete_product(product_id: int):
    await db.delete(product_id)

# WRITE-THROUGH - refresh the cache so the next GET is a HIT
@app.put(
    "/products/{product_id}",
    dependencies=[Depends(cache_put(eviction_group="products", key_builder=default_key_builder, ttl=300))],
)
async def replace_product(product_id: int, body: Product):
    return await db.update(product_id, body)

Cached responses include an X-Redis-Cache header (HIT, MISS, or BYPASS when the response was served but not stored) along with Cache-Control and ETag headers. If the response body depends on the user's session, pass vary_on_session=True to cache().

The example below drives a cached endpoint with FastAPI's TestClient so you can see the MISS → HIT → eviction cycle, plus the HTTP caching headers, in action:

# The first request is a MISS: the handler runs and the response is cached.
first = client.get("/cache-demo")
print(first.headers["X-Redis-Cache"])
# >>> MISS

# A second request within the TTL is a HIT, served from Redis without
# re-running the handler, so the cached body is returned unchanged.
second = client.get("/cache-demo")
print(second.headers["X-Redis-Cache"])
# >>> HIT

print(first.json() == second.json())
# >>> True

Evicting a resource clears its cached entry, so the following read is a MISS again:

# Deleting the resource evicts its cache entry, so the next read is a MISS
# again and the handler recomputes a fresh response.
client.delete("/cache-demo")

third = client.get("/cache-demo")
print(third.headers["X-Redis-Cache"])
# >>> MISS

print(third.json() == first.json())
# >>> False

CacheBackend for full control

For conditional caching, cascade invalidation, dynamic TTLs, or caching intermediate results, inject CacheBackendDep and call its get/set/delete/has/delete_group methods directly. Values are serialized to and from JSON automatically:

from redis_fastapi import CacheBackendDep

@app.get("/dashboard/{user_id}")
async def dashboard(user_id: int, cache: CacheBackendDep):
    cached = await cache.get(f"stats:{user_id}", eviction_group="dashboard")
    if cached is not None:
        return cached
    result = await compute_dashboard(user_id)
    await cache.set(f"stats:{user_id}", result, ttl=300, eviction_group="dashboard")
    return result

See the SDK caching guide for detailed patterns and best practices.

HTTP caching

Responses cached with the DI factories carry standard HTTP caching headers, so clients and proxies can revalidate cheaply. The SDK sets Cache-Control from the entry's TTL and emits a weak ETag; when a client returns that tag in an If-None-Match header, the server responds with 304 Not Modified and no body:

# Cached responses carry standard HTTP caching headers. Evict first so the next
# request is a fresh MISS whose Cache-Control reflects the full 30-second TTL.
client.delete("/cache-demo")
miss = client.get("/cache-demo")
print(miss.headers["Cache-Control"])
# >>> max-age=30

# The cached response also carries a weak ETag. Replaying it with If-None-Match
# lets the server answer 304 Not Modified with no body.
etag = miss.headers["ETag"]
not_modified = client.get("/cache-demo", headers={"If-None-Match": etag})
print(not_modified.status_code)
# >>> 304

Rate limiting

Add .rate_limiting() to the builder chain, then attach the rate_limit() dependency to a route. Counters live in Redis, so limits hold across every worker and pod. Stack two limits to cap both bursts and sustained traffic. Give each a distinct scope so their counters stay independent:

from fastapi import Depends, FastAPI
from redis_fastapi import FastAPIRedis, rate_limit

app = FastAPI()
FastAPIRedis(app).lifespan().rate_limiting()

@app.get(
    "/search",
    dependencies=[
        Depends(rate_limit("10/second", scope="search:burst")),
        Depends(rate_limit("100/minute", scope="search:sustained")),
    ],
)
async def search():
    return {"results": []}

Limits count per client IP by default. A request that exceeds either limit gets a 429 Too Many Requests response with a Retry-After header, and every response carries X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset headers.

See the SDK rate limiting guide for custom identifiers, an app-wide limiter, custom 429 responses, and the behavior when Redis is unreachable.

Sessions

Add .sessions() to the builder chain to keep session data in Redis. The cookie holds only an opaque session ID, and Redis enforces both an idle timeout and an absolute timeout. Inject SessionDep to read and write the session, and add valid_session() to return 401 when a request has no live session:

from fastapi import Depends, FastAPI
from redis_fastapi import FastAPIRedis, SessionDep, valid_session

app = FastAPI()
FastAPIRedis(app).lifespan().sessions()

@app.post("/login")
async def login(session: SessionDep):
    session["user_id"] = 42  # rotates the session ID after the endpoint completes
    return {"ok": True}

@app.get("/me", dependencies=[Depends(valid_session())])
async def me(session: SessionDep):
    return {"user_id": session.get("user_id")}

Changing the signed-in user rotates the session ID automatically. request.session also works, so code written for Starlette's SessionMiddleware runs unchanged. See the SDK sessions guide for timeouts, sign-out everywhere, CSRF (cross-site request forgery) protection, and encryption.

Cluster mode

To work with an OSS Cluster, set REDIS_CLUSTER=true and point REDIS_URL at the cluster nodes:

export REDIS_CLUSTER=true
export REDIS_URL=redis://node1:6379,node2:6379,node3:6379

In cluster mode, AsyncRedisDep yields an AsyncRedisCluster client.

Sentinel mode

To connect through Redis Sentinel, set REDIS_SENTINEL=true and list the Sentinels instead of setting REDIS_URL. The SDK asks the Sentinels for the address of the current primary, and asks again after a failover:

export REDIS_SENTINEL=true
export REDIS_SENTINEL_NODES=sentinel1:26379,sentinel2:26379,sentinel3:26379
export REDIS_SENTINEL_MASTER_NAME=mymaster

Your application code doesn't change. AsyncRedisDep yields a standard AsyncRedis client connected to the current primary. Sentinel mode can't be combined with REDIS_CLUSTER or REDIS_URL. If the Sentinels require authentication, set REDIS_SENTINEL_USERNAME and REDIS_SENTINEL_PASSWORD.

Observability

The SDK can emit OpenTelemetry spans and metrics for every cache operation. Telemetry is opt-in and a zero-cost no-op when disabled. Install the optional dependency and enable it on the builder:

pip install fastapi-redis-sdk[otel]

FastAPIRedis(app).lifespan().caching().otel()

Calling .otel() on the builder is what activates the cache telemetry. To also emit redis-py's low-level command spans and connection-pool metrics, set REDIS_OTEL_REDIS_ENABLED=true. See the SDK configuration guide for the full list of spans and metrics that are emitted.

More information

• fastapi-redis-sdk on GitHub
• fastapi-redis-sdk documentation
• fastapi-redis-sdk on PyPI
• redis-py client documentation
• FastAPI documentation