InMemoryCacheAdapter — In-Memory Cache Backend

A CacheProviderInterface implementation for tests and single-process usage, with no external Redis dependency. Locks are process-local.

InMemoryCacheAdapter

InMemoryCacheAdapter(
    key_prefix: str = "test:", cache_name: str = "memory"
)

Bases: CacheProviderInterface

Thread-safe in-memory cache implementation.

This adapter provides a complete cache implementation using Python dictionaries with thread-safe operations.

Features

Full CacheProviderInterface compliance
Thread-safe operations
TTL support with lazy expiration
Mock distributed locks (single-process only)

Example

cache = InMemoryCacheAdapter() cache.set("key", {"data": "value"}, ttl=timedelta(minutes=5)) data = cache.get("key") with cache.get_lock("my_lock") as lock: ... # Critical section (only within single process!) ... pass

Warning

This is for TESTING ONLY. Data is not persisted and locks only work within a single process.

Initialize in-memory cache.

Parameters:

Name	Type	Description	Default
`key_prefix`	`str`	Prefix for all cache keys	`'test:'`
`cache_name`	`str`	Name for metrics identification	`'memory'`

provider_name `property`

provider_name: str

Return 'memory' as the provider identifier.

cleanup_expired

cleanup_expired() -> int

Lock-acquiring entry point for periodic expiration sweeps.

Wraps the unlocked _cleanup_expired() with self._lock so external callers (CleanupService) can safely invoke it. self._lock is a non-reentrant threading.Lock; reusing the private method from a non-locked context would deadlock.

Returns:

Type	Description
`int`	Number of expired entries removed.

clear_all_instances `classmethod`

clear_all_instances() -> None

Clear the class-level weak registry (for test fixtures).

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 on a dict-valued record (lock-wrapped).

get_lock

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

Get a distributed lock instance.

Resolves the registry key once via self._make_key(name) so two adapter instances with different key_prefix map the same user-facing name to distinct entries in the class-level InMemoryLock._locks registry.

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.

push_limit

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

Append value to a list and trim to max_len under thread lock.

list_range

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

Return elements from start to end (inclusive).

health_check

health_check() -> bool

Check if cache is healthy.

set_health_status

set_health_status(healthy: bool) -> None

Set health status for testing.

flush_all

flush_all() -> bool

Clear all keys.

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.

get_store_size

get_store_size() -> int

Get number of entries in store (for testing).

clear_all

clear_all() -> None

Clear entire store including all prefixes (for testing cleanup).

InMemoryCacheAdapter — In-Memory Cache Backend