Official FastAPI integration for Redis

Idiomatic Redis integration for FastAPI - connection management and DI-based caching with automatic key consistency.

Features

• Fluent setup - FastAPIRedis(app).lifespan().caching() configures pools and caching in one chain
• DI-based caching - cache(), cache_evict(), cache_put() as Depends() factories, plus CacheBackend for complex invalidation and conditional logic
• HTTP-native caching - ETag, 304 Not Modified, Cache-Control directives out of the box
• Testable - full dependency_overrides support for caching; no monkey-patching
• Production-ready - TLS, auth, OSS Cluster mode, configurable via environment variables or .env

Requirements

Dependency	Supported versions
Python	3.10 to 3.14
FastAPI	0.115+
redis-py	6.0+
Pydantic	2.0+
Redis server	7.4+

Installation

pip install fastapi-redis-sdk

Quick Start

from fastapi import FastAPI
from redis_fastapi import FastAPIRedis, RedisDep, AsyncRedisDep

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

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

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

Connection pools are managed automatically and closed on shutdown. The builder wraps any existing lifespan - multiple libraries can each register their own without conflicting. See Configuration for connection options.

Caching

Two caching patterns for different needs, sharing the same connection pool:

Pattern	Best for
cache(), cache_evict(), cache_put()	Most endpoints - read/write/invalidate
CacheBackend	Complex invalidation, conditional logic

DI factories - read, invalidate, write-through

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

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


# READ - cache GET responses
@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 on delete
@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 - update 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)

All three factories share the same key_builder, so GET, DELETE, and PUT on the same path target the exact same cache key. Responses include X-Redis-Cache (HIT/MISS), Cache-Control, and ETag headers with 304 Not Modified support. Full dependency_overrides support makes testing straightforward - no monkey-patching required.

CacheBackend - full control via dependency injection

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:
        return cached
    result = await compute_dashboard(user_id)
    await cache.set(f"stats:{user_id}", result, ttl=300, eviction_group="dashboard")
    return result

Provides get/set/delete/has/delete_group with automatic JSON serialization. Use it for conditional caching, cascade invalidation, dynamic TTL, and intermediate result caching.

See the Caching Guide for detailed examples, feature comparison, and best practices.

Configuration

All settings are read from environment variables (prefixed REDIS_) or a .env file. Set REDIS_URL for the simplest setup:

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

Or configure individual fields:

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

Additional options: TLS (REDIS_SSL, REDIS_SSL_CERTFILE, etc.), connection pool (REDIS_MAX_CONNECTIONS, REDIS_SOCKET_TIMEOUT), OSS Cluster mode (REDIS_CLUSTER=true), key prefix (REDIS_PREFIX), and default cache TTL (REDIS_DEFAULT_TTL, default 0 = no expiry).

For programmatic configuration:

from redis_fastapi import get_settings

settings = get_settings()
settings.url = "redis://custom:6379/0"
settings.default_ttl = 120

See the Configuration Guide for the full environment variable reference and API details.