Skip to content

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 | None = None,
    socket_connect_timeout: float | None = None,
    retry_on_timeout: bool | None = None,
)

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 | None

Socket read timeout in seconds. When None (default), resolves from RedisSettings.socket_timeout via the connection factory (same source as the async adapter).

 None
socket_connect_timeout float | None

Socket connect timeout in seconds. When None (default), resolves from RedisSettings.socket_connect_timeout.

 None
retry_on_timeout bool | None

Retry-on-timeout flag. When None (default), resolves from RedisSettings.retry_on_timeout.

 None

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.

cas_takeover 

cas_takeover(
    key: str,
    new_record: dict[str, Any],
    *,
    stale_before: float,
    ttl: timedelta | None = None
) -> bool

Atomic failed / stale-executing takeover via one EVAL (single-winner).

Surfaces a cache outage (raises AdapterConnectionError) rather than swallowing to False — this is a dedup-gate-only acquire op whose failure must reach the gate's fail-open path, mirroring :meth:setnx.

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 every key under the configured prefix (not the entire Redis DB).

The prefix is Baldur's namespace rather than a cache-only sub-namespace, so this also deletes circuit breaker state, DLQ entries and idempotency records stored under it. Read the interface warning before calling this against a live deployment.

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

Async Redis adapter

An AsyncCacheProviderInterface implementation over redis.asyncio, backing the awaitable idempotency dedup gate so aprotect(idempotency_key=…) performs its cross-worker dedup via native await. Shares key-prefix and serialization with RedisCacheAdapter, so async and sync writes hit the same Redis keys.

AsyncRedisCacheAdapter 

AsyncRedisCacheAdapter(
    url: str | None = None,
    client: Any | None = None,
    key_prefix: str | None = None,
    socket_timeout: float | None = None,
    socket_connect_timeout: float | None = None,
    retry_on_timeout: bool | None = None,
)

Bases: AsyncCacheProviderInterface

Redis implementation of the async dedup cache surface (redis.asyncio).

Example

cache = AsyncRedisCacheAdapter() await cache.asetnx("k", {"status": "executing"}, ttl=timedelta(minutes=5)) True await cache.aclose() # drain the pool on shutdown

Initialize the async Redis cache adapter.

Parameters:

Name Type Description Default
url str | None

Redis URL. When None, resolves from BALDUR_REDIS_URL via get_redis_settings() (same source as the sync adapter).

 None
client Any | None

Pre-configured redis.asyncio client (takes precedence; used by unit tests to inject a fake).

 None
key_prefix str | None

Tri-state prefix selector, identical semantics to :class:RedisCacheAdapter — None (default) resolves the dynamic per-operation prefix (TestModeContext / NamespaceSettings aware) so async keys match the sync adapter's keys; "" adds no prefix; a literal string is a static override.

 None
socket_timeout float | None

Socket read timeout. None reads RedisSettings.socket_timeout — the same source the sync adapter resolves from, so both clients share one configuration.

 None
socket_connect_timeout float | None

Socket connect timeout. None reads RedisSettings.socket_connect_timeout.

 None
retry_on_timeout bool | None

Retry-on-timeout flag. None reads RedisSettings.retry_on_timeout.

 None

provider_name property 

provider_name: str

Return 'redis' as the provider identifier.

asetnx async 

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

Atomic SET NX (+ optional PX/EX) — single-round-trip acquire.

acas_dict_field async 

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

Reuses the sync adapter's Lua script (:data:LUA_CAS_DICT_FIELD) so the async and sync CAS semantics are byte-for-byte identical on shared keys.

acas_takeover async 

acas_takeover(
    key: str,
    new_record: dict[str, Any],
    *,
    stale_before: float,
    ttl: timedelta | None = None
) -> bool

Atomic failed / stale-executing takeover via one EVAL (single-winner).

Reuses the sync adapter's Lua script (:data:LUA_CAS_TAKEOVER) so the async and sync takeover semantics are byte-for-byte identical on shared keys. Surfaces a cache outage (raises AdapterConnectionError) rather than swallowing to False — this dedup-gate-only acquire op must reach the gate's fail-open path, mirroring :meth:asetnx.

aclose async 

aclose() -> None

Drain the redis.asyncio connection pool. Idempotent, best-effort.

Registered with the framework-independent GracefulShutdownCoordinator at resolution time (see the async cache resolver) so this second pool — separate from the sync RedisCacheAdapter pool — is drained on graceful shutdown / process recycle across every framework adapter (Django/Flask/FastAPI/CLI). This closes the dev hot-reload leak vector (uvicorn --reload recycling the process without draining the async pool).