Idiomatic Redis integration for FastAPI - connection management and DI-based caching with automatic key consistency.
- 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()asDepends()factories, plusCacheBackendfor complex invalidation and conditional logic - HTTP-native caching —
ETag,304 Not Modified,Cache-Controldirectives out of the box - Rate limiting —
rate_limit()dependency with a fluent rate language ("10/second"),X-RateLimit-*/Retry-Afterheaders, and distributed per-client counters - Testable — full
dependency_overridessupport; no need for monkey-patching - Pydantic-validated configuration — fully configurable via environment variables or via an
.envfile
| Dependency | Supported versions |
|---|---|
| Python | 3.10 to 3.14 |
| FastAPI | 0.115+ |
| redis-py | 6.0+ |
| Pydantic | 2.0+ |
| Redis server | 7.4+ |
pip install fastapi-redis-sdkCache GET responses with a dependency — on a hit the endpoint is skipped, on a miss the response is stored after it returns:
from fastapi import Depends, FastAPI
from redis_fastapi import FastAPIRedis, cache
app = FastAPI()
FastAPIRedis(app).lifespan().caching()
@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)cache_evict() and cache_put() handle invalidation and write-through with matching keys, and CacheBackendDep exposes imperative get/set/delete/has/delete_group for conditional logic. Cached responses carry X-Redis-Cache (HIT/MISS), Cache-Control, and ETag headers with 304 Not Modified support.
See the Caching Guide for the full patterns, CacheBackend usage, Pydantic model caching, feature comparison, and best practices.
Protect an endpoint with the rate_limit() dependency. Stack two limits to cap bursts and sustained traffic at once — a short high-rate window plus a longer low-rate window:
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")), # burst
Depends(rate_limit("100/minute", scope="search:sustained")), # sustained
],
)
async def search():
return {"results": [...]}Both limits count per client IP by default; a request must satisfy both, and the distinct scope keeps the two counters independent on the same route. When either is exceeded the request gets a 429 Too Many Requests with Retry-After, and every response carries X-RateLimit-Limit / -Remaining / -Reset. Counters live in Redis, so limits hold across every worker and pod.
See the Rate Limiting Guide for identifiers, the global limiter, custom responses, IETF headers, and the imperative backend.
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/0Or configure individual fields:
export REDIS_HOST=redis.example.com
export REDIS_PORT=6380
export REDIS_PASSWORD=secretAdditional 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 = 120See the Configuration Guide for the full environment variable reference and API details.