FastAPI Redis¶

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

Features¶

Fluent setup — FastAPIRedis(app).lifespan().caching() configures pools and caching in one chain, attaching to the FastAPI lifespan events
Dependency injection — 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; no need for monkey-patching
Pydantic-validated configuration — fully configurable via environment variables or via an .env file

Installation¶

pip install fastapi-redis-sdk

Requires Python 3.10+ and a Redis 7.4+ server.

See the Installation page for full requirements and alternative package managers.

Quick start¶

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")}

Connection pools are managed automatically and closed on shutdown. The builder wraps any existing lifespan — multiple libraries can each register their own without conflicting.

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

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.

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:

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

See the Configuration Guide for the full environment variable reference.