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, 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
• Rate limiting — rate_limit() dependency with a fluent rate language ("10/second"), X-RateLimit-* / Retry-After headers, and distributed per-client counters
• Testable — full dependency_overrides support; no need for monkey-patching
• Pydantic-validated configuration — fully configurable via environment variables or via an .env file

Requirements

Installation

pip install fastapi-redis-sdk

Caching

Cache 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.

Rate limiting

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.

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.