RedisCacheAdapter — Redis Cache Backend

The production-default CacheProviderInterface implementation. Backs distributed locks with Redlock-style primitives so cache coordination is safe across processes and hosts.

RedisCacheAdapter

RedisCacheAdapter(
    url: str | None = None,
    client: Any | None = None,
    key_prefix: str | None = None,
    default_ttl: timedelta | None = None,
    socket_timeout: float = 5.0,
    socket_connect_timeout: float = 5.0,
    retry_on_timeout: bool = True,
)

Bases: CacheProviderInterface

Redis implementation of CacheProviderInterface.

This adapter provides full Redis caching functionality including distributed locks, atomic counters, and TTL management.

Configuration

url can be passed explicitly. When omitted (no-arg construction), the adapter reads BALDUR_REDIS_URL via :func:get_redis_settings, matching RedisConnectionFactory's URL source.

Example

cache = RedisCacheAdapter(url="redis://localhost:6379/0") cache.set("key", {"data": "value"}, ttl=timedelta(minutes=5)) data = cache.get("key") with cache.get_lock("my_lock") as lock: ... # Critical section ... pass

Initialize Redis cache adapter.

Parameters:

Name	Type	Description	Default
`url`	`str \| None`	Redis URL (e.g., "redis://localhost:6379/0"). When `None`, resolves from `BALDUR_REDIS_URL` via `get_redis_settings()`.	`None`
`client`	`Any \| None`	Pre-configured Redis client (takes precedence over url)	`None`
`key_prefix`	`str \| None`	Tri-state prefix selector: `None` (default) — per-operation dynamic prefix via :func:`get_effective_key_prefix`. Honors :class:`TestModeContext` (`xtest:` synthetic isolation) and :class:`NamespaceSettings` (region/tenant/env separation). Symmetric with :class:`ResilientStorageBackend`. `""` — composer pattern: no prefix added. The caller is responsible for prepending its own prefix (`ResilientStorageBackend`, `rate_limit_tracker`). `"static:"` — static literal override. Used by tests for per-instance isolation that must NOT shift with TestModeContext.	`None`
`default_ttl`	`timedelta \| None`	Default TTL for set operations	`None`
`socket_timeout`	`float`	Socket timeout for operations	`5.0`
`socket_connect_timeout`	`float`	Socket connection timeout	`5.0`
`retry_on_timeout`	`bool`	Retry on timeout errors	`True`

provider_name `property`

provider_name: str

Return 'redis' as the provider identifier.

raw_client `property`

raw_client: Any

Return the underlying redis client.

Public seam for composed components that need the raw client for operations the cache interface does not expose (e.g. Lua eval, pipelines). The raw client already legitimately escapes the adapter via :meth:get_lock, which hands it to DistributedLock.

get

get(key: str) -> Any | None

Get value by key.

set

set(
    key: str, value: Any, ttl: timedelta | None = None
) -> bool

Set value with optional TTL.

delete

delete(key: str) -> bool

Delete key from cache.

exists

exists(key: str) -> bool

Check if key exists in cache.

incr

incr(key: str, amount: int = 1) -> int

Atomically increment a counter.

decr

decr(key: str, amount: int = 1) -> int

Atomically decrement a counter.

expire

expire(key: str, ttl: timedelta) -> bool

Set expiration on existing key.

ttl

ttl(key: str) -> int | None

Get remaining TTL in seconds.

setnx

setnx(
    key: str, value: Any, ttl: timedelta | None = None
) -> bool

Set value only if key does not exist.

cas_dict_field

cas_dict_field(
    key: str,
    field: str,
    expected: Any,
    new_value: dict[str, Any],
    ttl: timedelta | None = None,
) -> bool

Atomic single-field CAS via cjson.decode + SET PX in one EVAL.

get_lock

get_lock(
    name: str,
    timeout: timedelta = timedelta(seconds=10),
    blocking_timeout: float | None = None,
) -> DistributedLock

Get a distributed lock instance.

Resolves the storage key once via self._make_key(name) and passes the result to the lock constructor as full_key. The lock snapshots the resolved key for its lifetime — acquire / release / extend operate on the same Redis key regardless of TestModeContext boundary crossings.

mget

mget(keys: list[str]) -> dict[str, Any]

Get multiple values at once.

mset

mset(
    mapping: dict[str, Any], ttl: timedelta | None = None
) -> bool

Set multiple values at once.

mdelete

mdelete(keys: list[str]) -> int

Delete multiple keys at once.

hget

hget(name: str, key: str) -> Any | None

Get a field from a hash.

hset

hset(name: str, key: str, value: Any) -> bool

Set a field in a hash.

hgetall

hgetall(name: str) -> dict[str, Any]

Get all fields from a hash.

push_limit

push_limit(
    key: str,
    value: Any,
    max_len: int,
    ttl: timedelta | None = None,
) -> int

Atomically append value and trim list to max_len via RPUSH+LTRIM+EXPIRE.

list_range

list_range(key: str, start: int, end: int) -> list[Any]

Return elements from start to end (inclusive) via LRANGE.

health_check

health_check() -> bool

Check if Redis is reachable.

flush_all

flush_all() -> bool

Clear all keys with our prefix (not entire Redis DB).

keys

keys(pattern: str = '*') -> list[str]

Find keys matching a pattern.

scan

scan(
    pattern: str = "*", count: int = 100
) -> tuple[int, list[str]]

Incrementally iterate keys matching a pattern.

close

close() -> None

Disconnect the Redis connection pool. Idempotent.

Drains all sockets held by the underlying redis-py connection pool. Required by the test-fixture reset chain: reset_init_state() re-runs init() repeatedly under xdist, and without an explicit pool drain each iteration leaks file descriptors until the runner trips "too many open files".

reconnect

reconnect() -> bool

Reset the connection pool - release dead connections and reconnect.

redis-py's ConnectionPool.disconnect() closes all connections in the pool. A subsequent ping() call causes the pool to create new connections automatically.

Returns:

Type	Description
`bool`	Whether reconnection succeeded

RedisCacheAdapter — Redis Cache Backend