Redis Fastapi Redis Sdk Fastapi Redis Sdk

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      # Tests on a specific Python version
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).
• 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.

Code style

• ruff for linting + formatting (line-length 88).
• mypy strict mode, Python 3.10 target.
• from __future__ import annotations in all source files.
• Type aliases use Union[] (not X | Y) for Python 3.10 compat in deps.py and types.py (ruff rule UP007 ignored there).

CI

CI is defined in .github/workflows/ci.yml and delegates to nox sessions. Do NOT add inline uv run ruff / uv run mypy commands to CI — use uv run nox -s <session> so CI and local checks stay in sync.

Do NOT change

• Build system. The project uses uv_build as its build backend. NEVER replace it with hatchling, setuptools, flit, or any other build backend. All build config lives in pyproject.toml under [build-system] and [tool.uv.build-backend].
• DI approach. Caching (for example) is implemented as Depends() factories (cache(), cache_evict(), cache_put()), NOT as decorators. NEVER refactor DI based solutions to use a decorator-based approach (@cache). The DI pattern is a deliberate design decision, see /guide/architecture.md for more information.

Cache key format

Default key: {prefix}:{eviction_group}:{path}:{sorted_query_params}. Eviction group is wrapped in hash-tag braces {ns} for Redis Cluster slot alignment. Query params are sorted alphabetically for determinism. Headers are NOT part of the key — use a custom key_builder for header-dependent responses (e.g. Accept).

TTL behavior

Fixed-window expiry. Accessing a cached entry does NOT extend its TTL. ttl=0 or ttl=None means no automatic expiration.

Error handling

All Redis errors (RedisError, OSError) are caught and logged as warnings. Cache reads return None/miss; writes are silently dropped. Exceptions from endpoints are never cached — only successful responses. Corrupted cache data is treated as a miss (auto-fallback).

Common pitfalls

• anyio.from_thread.run takes a zero-arg callable returning an awaitable, NOT a coroutine object. Wrap with lambda.
• SyncCacheBackendDep must be imported at module level when using from __future__ import annotations, or FastAPI cannot resolve the type.
• Do not use uv add --dev in CI workflows — deps are managed by nox sessions and pyproject.toml dependency groups.
• cache() reads get_settings() at dependency-creation time (when the module loads), not per-request. Runtime setting changes won't be picked up by already-registered routes.

Source: redis/fastapi-redis-sdk — distributed by TomeVault.