FastAPI-Redis SDK Development

fastapi-redis-sdk

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

Project layout

src/redis_fastapi/       # Library source (single flat package)
  __init__.py            # Public API re-exports
  setup.py               # FastAPIRedis fluent builder
  lifespan.py            # redis_lifespan async context manager
  deps.py                # FastAPI DI factories & type aliases
  config.py              # RedisSettings (pydantic-settings)
  cache.py               # cache(), cache_evict(), cache_put() + middleware
  cache_backend.py       # CacheBackend (async) + SyncCacheBackend
  telemetry.py           # Optional OpenTelemetry instrumentation
  types.py               # Coder, KeyBuilder protocols
tests/
  unit/                  # fakeredis-based, no real Redis needed
  integration/           # Requires Redis on localhost:6379
pyproject.toml           # uv build backend, dependency groups, tool config
noxfile.py               # CI-mirroring sessions (lint, typecheck, security, tests, docs)

Setup and tooling

Package manager: uv. Task runner: nox (with uv venv backend).

uv sync --all-groups          # Install all deps
uv run nox                    # Run ALL CI checks locally
uv run nox -s lint            # Lint + format check only
uv run nox -s typecheck       # mypy
uv run nox -s tests-3.12      # Full suite on a specific Python version
uv run nox -s tests_unit-3.12 # Unit suite only (fakeredis, no server)
uv run nox -s tests_integration-3.12  # Integration suite only (needs Redis)
uv run nox -s fix             # Auto-fix lint/format
uv run nox -s docs_serve      # Live-reload docs at localhost:8000

Architecture rules

Lifespan is mandatory

Redis pools MUST be initialised via the app lifespan. There is no fallback. Accessing the pool without a lifespan raises RuntimeError.

app = FastAPI()
FastAPIRedis(app).lifespan().caching()  # Always call .lifespan()

Async-first, sync via bridge

• All DI factories (get_async_redis, get_cache_backend) are async.
• FastAPI runs them correctly even for sync endpoints (threadpool).
• SyncCacheBackend wraps async calls via anyio.from_thread.run — use only from sync endpoints running in FastAPI's worker threads.

Dependency injection types

from redis_fastapi import AsyncRedisDep, CacheBackendDep, SyncCacheBackendDep

# Async endpoint — use AsyncRedisDep or CacheBackendDep
async def endpoint(redis: AsyncRedisDep): ...
async def endpoint(cache: CacheBackendDep): ...

# Sync endpoint — use SyncCacheBackendDep
def endpoint(cache: SyncCacheBackendDep): ...

Lifespan wrapping

.lifespan() wraps the app's existing lifespan — it does not replace it. Multiple builder calls nest around whatever is already there. For explicit ordering, skip .lifespan() and compose manually with redis_lifespan:

from redis_fastapi import FastAPIRedis, redis_lifespan

@asynccontextmanager
async def my_lifespan(app):
    async with redis_lifespan(app):
        async with db_lifespan(app):
            yield

app = FastAPI(lifespan=my_lifespan)
FastAPIRedis(app).caching()   # no .lifespan() — user owns it

See docs/guide/architecture.md § Lifespan wrapping for details.

Caching patterns

Two patterns, same pool:

1. DI factories — cache(ttl=N, eviction_group="x"), cache_evict(...), cache_put(...) as Depends(). Requires .caching() on setup.
   • Automatic HTTP semantics (ETag, 304, Cache-Control, X-Redis-Cache).
   • Only GET requests are cached; non-GET bypasses cache entirely.
   • Graceful degradation: Redis failures log warnings, never crash.
2. CacheBackend — get/set/delete/has/delete_group. For conditional logic, cascade invalidation, dynamic TTL.
   • Accepts timedelta for TTL (DI factories accept int seconds only).
   • No automatic HTTP headers — add manually if needed.

Choose cache() for most GET endpoints. Choose CacheBackend when you need conditional caching, multi-step invalidation, or custom serializers. cache_evict()/cache_put() bridge writes back to the same cache keys.

Cache hit/miss internals

• Hit: DI dependency raises CacheHitException → exception handler returns cached response. Endpoint never executes.
• Miss: CacheResponseCaptureMiddleware buffers the response body and stores it in Redis after the endpoint returns.
• No full ASGI middleware for reads — benchmarks showed no gain over DI. See docs/guide/architecture.md § Why not a full ASGI middleware.

Storage model

Cached entries are Redis string keys with eviction-group prefixes. Namespace deletion uses SCAN + DEL. Hash-based storage (faster eviction-group deletion via single DEL) is a future opt-in gated on Redis ≥ 8.0. See docs/guide/architecture.md § Storage model.

Telemetry

Three independent OTel layers (each opt-in):

1. HTTP spans — opentelemetry-instrumentation-fastapi (external).
2. Cache operation spans + metrics — FastAPIRedis(app)...otel() or REDIS_OTEL_ENABLED=true. Install fastapi-redis-sdk[otel].
3. Redis command spans — REDIS_OTEL_REDIS_ENABLED=true or opentelemetry-instrumentation-redis (not both).

See docs/guide/architecture.md § Telemetry for span names and metrics.

Testing conventions

• Unit tests (tests/unit/): use fakeredis.aioredis, no real Redis.
• Integration tests (tests/integration/): real Redis, decorated with @requires_redis (auto-skip if server unreachable).
• filterwarnings = ["error"] in pytest config — all warnings are errors.
• Coverage threshold: 80% (--cov-fail-under=80). The integration-only session lowers this to 70 (_INTEGRATION_COV_FLOOR in noxfile.py) — 83 integration tests alone cover ~78% of src/.
• CI runs the two suites in separate jobs: unit-tests runs tests_unit across 3 OSes x 5 Pythons, integration-tests runs tests_integration against Redis 8.8 and 7.4. See docs/guide/compatibility.md § 4.
• asyncio_mode = "auto" — no need for @pytest.mark.asyncio.

Test fixtures

• fake_async_redis — fakeredis instance, available in unit tests.
• real_redis / real_async_redis — real Redis clients for integration.
• Every integration test flushes the DB on teardown.

Configuration

All settings via env vars prefixed REDIS_ or .env file. Key vars: REDIS_URL, REDIS_HOST, REDIS_PORT, REDIS_PASSWORD, REDIS_SSL, REDIS_CLUSTER, REDIS_PREFIX, REDIS_DEFAULT_TTL.

Sentinel mode: REDIS_SENTINEL=true, REDIS_SENTINEL_NODES (comma-separated host:port), REDIS_SENTINEL_MASTER_NAME. The primary's pool is a SentinelConnectionPool in _PoolState.async_pool; _PoolState.async_sentinel holds the Sentinel clients, which the lifespan closes.