API Reference

Protocols

AsyncStringSerializerProtocol

source

Protocol for value serialization.

Implementations handle conversion between Python objects and string representations suitable for storage or transmission. Used across extensions (cache, messaging, tasks, etc.) for consistent serialization.

Example

class JSONSerializer:
    async def serialize(self, value: Any) -> str:
        return json.dumps(value)

    async def deserialize(self, data: str) -> Any:
        return json.loads(data)

class JSONSerializer:
    async def serialize(self, value: Any) -> str:
        return json.dumps(value)

    async def deserialize(self, data: str) -> Any:
        return json.loads(data)

serialize

async def serialize(value: Any) -> str

source

Serialize a Python object to string.

Parameters

Parameter	Type	Description
value	Any	Python object to serialize.

Returns

Type	Description
str	String representation.

deserialize

async def deserialize(data: str) -> Any

source

Deserialize string back to Python object.

Parameters

Parameter	Type	Description
data	str	Serialized string data.

Returns

Type	Description
Any	Reconstructed Python object.

---

CacheAdminProtocol

source

Optional operational contract exposed by inspectable cache backends.

CacheBackendProtocol intentionally stays focused on cache data-plane operations. Operator consoles also need a safe, backend-neutral way to inspect and reset counters without reaching into private implementation fields. Backends that can expose local counters implement this optional protocol; distributed backends may return None for unavailable statistics.

get_stats

def get_stats() -> dict[str, int | float | str] | None

source

Return backend counters and capacity information, when available.

reset_stats

def reset_stats() -> None

source

Reset operator-facing counters without deleting cached values.

---

CacheBackendProtocol

source

Protocol for cache backend implementations.

This interface defines the contract that all cache backend implementations must follow. It provides a unified API for different storage mechanisms (memory, Redis, Memcached, etc.).

Example

class RedisBackend:
    async def get(self, key: str) -> Result[Any | None, CacheError]:
        try:
            data = await self._redis.get(key)
            return Ok(self._serializer.deserialize(data) if data else None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

    async def set(
        self, key: str, value: Any, ttl: int | None = None
    ) -> Result[None, CacheError]:
        try:
            data = self._serializer.serialize(value)
            await self._redis.set(key, data, ex=ttl)
            return Ok(None)
        except Exception as exc:
            return Err(CacheError(str(exc)))

get

async def get(key: str) -> Result[Any | None, CacheError]

source

Get a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to retrieve.

Returns

Type	Description
Result[Any | None, CacheError]	Ok(value) if found, Ok(None) if not found, Err(CacheError) on failure.

Note

The return type is Any by design: the cache is a heterogeneous store and the caller knows the expected type. Use cast(T, result.unwrap()) at call sites for type safety.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
key	str	The cache key to set.
value	Any	The value to cache.
ttl	int | None	Time to live in seconds (optional).

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

delete

async def delete(key: str) -> Result[bool, CacheError]

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to delete.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure.

delete_many

async def delete_many(keys: list[str]) -> Result[int, CacheError]

source

Delete multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to delete.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

delete_pattern

async def delete_pattern(pattern: str) -> Result[int, CacheError]

source

Delete all keys matching a glob-style pattern.

Parameters

Parameter	Type	Description
pattern	str	Glob pattern (e.g. `"pet:list:*"). Supports *` as a wildcard matching any sequence of characters.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

exists

async def exists(key: str) -> Result[bool, CacheError]

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to check.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure.

clear

async def clear() -> Result[None, CacheError]

source

Clear all values from the cache.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

get_many

async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to retrieve.

Returns

Type	Description
Result[dict[str, Any], CacheError]	Ok(dict) mapping found keys to values, Err(CacheError) on failure.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set multiple values in the cache.

Parameters

Parameter	Type	Description
items	dict[str, Any]	Dictionary of key-value pairs to cache.
ttl	int | None	Time to live in seconds for all items.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if all items set successfully, Err(CacheError) on failure.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured HealthCheckResult.

---

CacheHealthCheckerProtocol

source

Protocol for cache health checking.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache.

Returns

Type	Description
HealthCheckResult	Structured HealthCheckResult.

---

CacheKeyBuilderProtocol

source

Protocol for building cache keys from function arguments.

build_key

def build_key(
    func: Callable[Ellipsis, Any],
    args: tuple[Any, Ellipsis],
    kwargs: dict[str, Any],
    prefix: str | None = None
) -> str

source

Build a cache key from function call information.

Parameters

Parameter	Type	Description
func	Callable[Ellipsis, Any]	Function being cached.
args	tuple[Any, Ellipsis]	Positional arguments.
kwargs	dict[str, Any]	Keyword arguments.
prefix	str | None	Optional key prefix.

Returns

Type	Description
str	Unique cache key string.

---

CacheProtectionStrategyProtocol

source

Protocol for cache stampede protection strategies.

acquire_lock

async def acquire_lock(
    key: str,
    ttl: int
) -> bool

source

Attempt to acquire a lock for cache population.

Parameters

Parameter	Type	Description
key	str	Cache key to lock.
ttl	int	Lock TTL in seconds.

Returns

Type	Description
bool	True if lock was acquired.

release_lock

async def release_lock(key: str) -> bool

source

Release a previously acquired lock.

Parameters

Parameter	Type	Description
key	str	Cache key to unlock.

Returns

Type	Description
bool	True if released successfully.

wait_for_value

async def wait_for_value(
    key: str,
    timeout: float,
    check_interval: float = 0.1
) -> Any | None

source

Wait for a value to be populated.

Parameters

Parameter	Type	Description
key	str	Cache key to wait for.
timeout	float	Maximum wait time in seconds.
check_interval	float	Check interval in seconds.

Returns

Type	Description
Any | None	Cached value if found, None if timeout.

---

CacheStatusHandler

source

Protocol for cache status handlers.

Implement this protocol to create custom handlers for cache status events. Handlers are registered with the cache status registry and called when corresponding cache operations occur.

can_handle

def can_handle(status: CacheStatus) -> bool

source

Check if this handler can handle the status.

Parameters

Parameter	Type	Description
status	CacheStatus	The cache status to check.

Returns

Type	Description
bool	True if this handler can handle the status.

record

async def record(
    metrics: CacheMetrics,
    count: int
) -> None

source

Record metrics for the status.

Parameters

Parameter	Type	Description
metrics	CacheMetrics	The metrics object to update.
count	int	The number of operations to record.

---

Classes

AutoRenewingLock

source

Distributed lock with automatic TTL renewal.

Prevents lock expiration during long operations by automatically extending the TTL in the background.

__init__

def __init__(
    redis: Any,
    key: str,
    lock_id: str,
    ttl: int,
    auto_renew: bool = True,
    renew_interval: int | None = None
) -> None

source

Initialize auto-renewing lock.

Parameters

Parameter	Type	Description
redis	Any	Redis client
key	str	Lock key
lock_id	str	Unique lock identifier (owner ID)
ttl	int	Lock TTL in seconds
auto_renew	bool	Enable automatic renewal
renew_interval	int | None	Renewal interval (default: ttl / 3)

start_auto_renewal

async def start_auto_renewal() -> None

source

Start background task to auto-renew lock.

stop_auto_renewal

async def stop_auto_renewal() -> None

source

Stop background renewal task.

release

async def release() -> None

source

Release lock and stop auto-renewal.

---

BackendType

source

Supported cache backend types.

---

CacheBackendConfig

source

Unified (flattened-union) configuration for any single cache backend.

This class uses the flattened-union pattern: all backend-specific fields live in one dataclass and are partitioned by the type field. Fields that do not belong to the active backend type are silently ignored by the respective backend implementation.

When to use

Prefer this class when loading backend configuration from environment variables or a flat config file (e.g. TOML/YAML with no nested sections), where per-backend types cannot be selected at parse time.

For strongly-typed, up-front configuration prefer the dedicated classes: MemoryBackendConfig, RedisBackendConfig, or MemcachedBackendConfig.

Field partitioning

+-------------------------+----------+----------+-----------+ | Field | memory | redis | memcached | +=========================+==========+==========+===========+ | name, type, default, | ✓ | ✓ | ✓ | | enabled, default_ttl, | | | | | key_prefix, | | | | | enable_metrics, | | | | | health_check_interval | | | | +-------------------------+----------+----------+-----------+ | max_size, | ✓ | | | | cleanup_interval | | | | +-------------------------+----------+----------+-----------+ | redis_host/port/db/ | | ✓ | | | password/url/ssl/ | | | | | pool_size | | | | +-------------------------+----------+----------+-----------+ | memcached_host/port/ | | | ✓ | | servers | | | | +-------------------------+----------+----------+-----------+

Attributes: name: Unique name for this backend. type: Backend type (memory, redis, memcached). default: Whether this is the default backend. enabled: Whether this backend is enabled. default_ttl: Default TTL in seconds. key_prefix: Key prefix for namespace isolation. enable_metrics: Enable metrics collection. health_check_interval: Health check interval in seconds.

build_redis_url

def build_redis_url() -> CacheBackendConfig

source

Build Redis URL from components if not provided.

build_memcached_servers

def build_memcached_servers() -> CacheBackendConfig

source

Build Memcached server list if not provided.

---

CacheConfig

source

Top-level configuration for Quadkit Cache.

Attributes: name: Configuration name (default: “cache”) version: Config version enabled: Whether cache is enabled backends: List of cache backend configurations service: Cache service settings environment: Environment name debug: Debug mode

validate_production_security

def validate_production_security() -> CacheConfig

source

Block insecure cache configurations in production.

validate_default_backend

def validate_default_backend(
    cls,
    v: list[CacheBackendConfig]
) -> list[CacheBackendConfig]

source

Ensure exactly one default backend is specified.

validate_unique_names

def validate_unique_names(
    cls,
    v: list[CacheBackendConfig]
) -> list[CacheBackendConfig]

source

Ensure backend names are unique.

get_default_backend

def get_default_backend() -> CacheBackendConfig | None

source

Get the default backend configuration.

Returns

Type	Description
CacheBackendConfig | None	The first backend marked as default, or `None` if none exists.

get_backend

def get_backend(name: str) -> CacheBackendConfig | None

source

Get a backend configuration by name.

Parameters

Parameter	Type	Description
name	str	The backend name to look up.

Returns

Type	Description
CacheBackendConfig | None	Matching CacheBackendConfig, or `None` if not found.

get_provider_class

def get_provider_class(cls) -> type

source

Return the provider class for this config.

Returns

Type	Description
type	The CacheProvider class.

---

CacheConnectedEvent

source

Cache backend connection established.

---

CacheConnectedHook

source

Payload fired when the cache backend establishes its connection.

Attributes: backend: Identifier of the backend that connected (e.g. "redis").

---

CacheDecorator

source

Advanced caching decorator with multiple strategies.

Provides a class-based decorator with more configuration options and caching strategies.

__init__

def __init__(
    service: CacheService,
    strategy: str = 'remember',
    key_prefix: str = '',
    ttl: int | None = None,
    backend: str | None = None,
    protect: bool = True,
    condition: Callable[[Any], bool] | None = None
)

source

Initialize the cache decorator.

Parameters

Parameter	Type	Description
service	CacheService	CacheService instance
strategy	str	Caching strategy ("remember", "cache", "conditional")
key_prefix	str	Prefix for cache keys
ttl	int | None	Time-to-live in seconds
backend	str | None	Backend name
protect	bool	Enable stampede protection
condition	Callable[[Any], bool] | None	Condition function for conditional caching

---

CacheDisconnectedHook

source

Payload fired when the cache backend disconnects or is shut down.

Attributes: backend: Identifier of the backend that disconnected.

---

CacheEntry

source

Cache entry with metadata for stampede protection.

Attributes: value: Cached value. cached_at: When value was cached. expires_at: When value expires.

is_expired

property is_expired() -> bool

source

Check if entry has expired.

---

CacheEntryEvictedHook

source

Payload fired when a cache entry is evicted (TTL expiry or explicit removal).

Attributes: key: The cache key that was evicted. backend: Identifier of the backend that performed the eviction.

---

CacheEvictedEvent

source

Cache entry was evicted.

---

CacheHealthResult

source

is_healthy

property is_healthy() -> bool

source

to_dict

def to_dict() -> dict[str, Any]

source

---

CacheHitEvent

source

Cache lookup resulted in a hit.

---

CacheItem

source

Represents a cached item with metadata.

is_expired

property is_expired() -> bool

source

expires_at

property expires_at() -> datetime | None

source

touch

def touch() -> None

source

---

CacheMetrics

source

Cache performance metrics with atomic updates.

record

async def record(
    status: CacheStatus,
    latency: float = 0.0,
    count: int = 1
) -> None

source

Atomic record helper.

record_hit

async def record_hit(count: int = 1) -> None

source

Convenience method to record hits.

record_miss

async def record_miss(count: int = 1) -> None

source

Convenience method to record misses.

record_set

async def record_set(count: int = 1) -> None

source

Convenience method to record sets.

record_delete

async def record_delete(count: int = 1) -> None

source

Convenience method to record deletes.

record_error

async def record_error(count: int = 1) -> None

source

Convenience method to record errors.

reset

def reset() -> None

source

Reset counters while retaining the cache entries themselves.

hit_rate

property hit_rate() -> float

source

to_dict

async def to_dict() -> dict[str, Any]

source

---

CacheMissEvent

source

Cache lookup resulted in a miss.

---

CacheModule

source

Redis and in-memory cache backends with stampede protection.

Call configure to register a configured CacheProvider and expose CacheBackendProtocol for injection.

Usage

from quadkit.cache.config import CacheConfig

@module(
    imports=[CacheModule.configure(CacheConfig(...))]
)
class AppModule(Module):
    pass

from quadkit.cache.config import CacheConfig

@module(
    imports=[CacheModule.configure(CacheConfig(...))]
)
class AppModule(Module):
    pass

configure

def configure(
    cls,
    config: CacheConfig | dict[str, Any] | None = None
) -> DynamicModule

source

Create a CacheModule with explicit configuration.

Parameters

Parameter	Type	Description
config	CacheConfig | dict[str, Any] | None	CacheConfig, a plain `dict of config values, or None for framework defaults (or the orchestrator-injected cache` yaml section).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return an in-memory CacheModule for unit testing.

Uses an in-memory cache backend with no external Redis connection. SemanticCacheProtocol is not registered (not available without embedding client and vector index).

Returns

Type	Description
DynamicModule	A DynamicModule backed by in-memory cache storage.

---

CacheOperationConfig

source

Internal configuration for cache operations. Shared across backends.

Attributes: default_ttl: Default time-to-live in seconds. max_memory: Maximum memory usage in bytes. key_prefix: Prefix for all cache keys. enable_metrics: Whether to enable metrics collection. serializer_type: Type of serializer to use (json, msgpack).

make_key

def make_key(key: str) -> str

source

Create a prefixed cache key.

Parameters

Parameter	Type	Description
key	str	The raw cache key.

Returns

Type	Description
str	Key with the configured prefix applied, or the original key if no prefix is set.

strip_prefix

def strip_prefix(prefixed_key: str) -> str

source

Remove prefix from a cache key.

Parameters

Parameter	Type	Description
prefixed_key	str	A key that may carry the configured prefix.

Returns

Type	Description
str	The key with the prefix removed, or the original key unchanged.

---

CacheProvider

source

Quadkit Cache provider for quadkit integration.

Integrates cache services with quadkit’s provider system, managing lifecycle, configuration, and service registration.

__init__

def __init__(config: CacheConfig | None = None) -> None

source

Initialize the cache provider.

Parameters

Parameter	Type	Description
config	CacheConfig | None	Optional cache configuration. When provided the provider is configured immediately (equivalent to calling configure).

from_config

def from_config(
    cls,
    config: CacheConfig,
    **context: Any
) -> CacheProvider

source

Create a CacheProvider from a CacheConfig.

The provider is created and immediately configured with the config dict.

configure

def configure(config: dict[str, Any] | CacheConfig) -> None

source

Configure the provider with cache settings.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Start the cache provider.

All real I/O initialization (backend connections, protection setup, service wiring) happens here, after the container is frozen.

Parameters

Parameter	Type	Description
container	ContainerResolverProtocol	The DI container provided by the framework.

observe_repository

def observe_repository(
    cache_repository: Any,
    repository: Any
) -> None

source

Queue cache repository observation wiring for provider boot.

Parameters

Parameter	Type	Description
cache_repository	Any	Cache repository that exposes `observe()`.
repository	Any	Source repository whose mutations should invalidate cache.

shutdown

async def shutdown() -> None

source

Shutdown the cache provider and cleanup resources.

cache

property cache() -> CacheService

source

Get the default cache service.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check provider health.

get_health_status

async def get_health_status() -> HealthCheckResult

source

Return comprehensive health status of all cache services and backends.

Returns

Type	Description
HealthCheckResult	Structured HealthCheckResult.

get_metrics

async def get_metrics() -> dict[str, Any]

source

Return comprehensive metrics from all registered cache services.

Returns

Type	Description
dict[str, Any]	Metrics dictionary with per-service statistics.

---

CacheResult

source

Result of a cache operation.

is_hit

property is_hit() -> bool

source

is_miss

property is_miss() -> bool

source

hit

def hit(
    cls,
    value: T,
    key: str | None = None,
    latency: float | None = None,
    backend: str | None = None
) -> CacheResult[T]

source

miss

def miss(
    cls,
    key: str | None = None,
    latency: float | None = None
) -> CacheResult[T]

source

---

CacheService

source

High-level cache service with advanced features.

Provides a unified interface for caching operations across backend implementations (memory, Redis, Memcached). Stampede protection, metrics tracking, tag-based invalidation, and batch operations are composed in via mixin inheritance.

Attributes: provider: The cache backend provider. config: Configuration for cache operations.

__init__

def __init__(
    provider: CacheProvider | None = None,
    config: CacheOperationConfig | None = None,
    protection: StampedeProtectedCache | None = None,
    backend: CacheBackendProtocol | None = None,
    ctx: Context | None = None,
    max_tags: int = 10000
)

source

Initialize the cache service.

Parameters

Parameter	Type	Description
provider	CacheProvider | None	Cache provider for backend management (optional)
config	CacheOperationConfig | None	Service-level configuration
protection	StampedeProtectedCache | None	Cache stampede protection instance
backend	CacheBackendProtocol | None	Direct backend instance (optional, for standalone usage)
ctx	Context | None	Optional context for request-scoped cache keys
max_tags	int	Maximum number of tags tracked in the in-memory index. When the cap is reached, the oldest tag (by insertion/access order) is evicted. Defaults to 10 000.

close

async def close() -> None

source

Close the cache service, releasing any background resources.

get

async def get(
    key: str,
    backend: str | None = None,
    default: Any = None,
    request_scoped: bool = False
) -> Any

source

Get a value from cache with error handling.

Parameters

Parameter	Type	Description
key	str	Cache key
backend	str | None	Backend name (uses default if None)
default	Any	Default value if key not found

Returns

Type	Description
Any	Cached value or default

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None,
    backend: str | None = None
) -> bool

source

Set a value in cache with error handling.

Parameters

Parameter	Type	Description
key	str	Cache key
value	Any	Value to cache
ttl	int | None	Time-to-live in seconds
backend	str | None	Backend name (uses default if None)

Returns

Type	Description
bool	True if successful, False otherwise

delete

async def delete(
    key: str,
    backend: str | None = None
) -> bool

source

Delete a value from cache.

Parameters

Parameter	Type	Description
key	str	Cache key
backend	str | None	Backend name (uses default if None)

Returns

Type	Description
bool	True if successful, False otherwise

delete_pattern

async def delete_pattern(
    pattern: str,
    backend: str | None = None
) -> int

source

Delete all keys matching a glob-style pattern.

Delegates to the backend’s delete_pattern implementation. Redis uses SCAN + DEL; Memory iterates its in-process store. The namespace prefix (if any) is applied by the backend itself.

Parameters

Parameter	Type	Description
pattern	str	Glob pattern, e.g. `"pet:list:*"`.
backend	str | None	Named backend to use (uses default if `None`).

Returns

Type	Description
int	Number of keys deleted, or 0 on error.

exists

async def exists(
    key: str,
    backend: str | None = None
) -> bool

source

Check if a key exists in cache.

Parameters

Parameter	Type	Description
key	str	Cache key
backend	str | None	Backend name (uses default if None)

Returns

Type	Description
bool	True if key exists, False otherwise

clear

async def clear(backend: str | None = None) -> bool

source

Clear all values from cache.

Parameters

Parameter	Type	Description
backend	str | None	Backend name (uses default if None)

Returns

Type	Description
bool	True if successful, False otherwise

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return this service’s backend health and local metrics.

A provider aggregates the health of every service and backend. A service must therefore probe its backend directly rather than calling provider.health_check(): doing the latter re-enters the provider’s aggregation loop and ultimately leaves non-serializable health objects in a readiness response.

get_metrics

def get_metrics() -> dict[str, int]

source

Get service performance metrics.

Returns

Type	Description
dict[str, int]	Dictionary of metrics

backend

property backend() -> CacheBackendProtocol | None

source

Return the backing store, or None when only a provider is set.

Admin surfaces report per-cache occupancy and eviction from the backend itself; exposing it keeps that reporting out of the private attribute layout.

key_prefix

property key_prefix() -> str

source

Return this service’s key prefix ("" when it has none).

default_ttl

property default_ttl() -> int | None

source

Return the default TTL in seconds applied when set omits one.

get_backend_stats

def get_backend_stats() -> dict[str, int]

source

Return the backend’s own counters for administrative reporting.

get_stats is optional on the backend protocol — a Redis backend answers through INFO instead — so an unreporting backend yields zeros rather than fabricated activity.

Returns

Type	Description
dict[str, int]	Mapping with `hits, misses, evictions and entries`.

reset_metrics

def reset_metrics() -> None

source

Reset performance metrics.

get_typed

async def get_typed(
    key: str,
    type_: type[_T],
    backend: str | None = None,
    default: _T | None = None
) -> _T | None

source

Get a value from cache and cast it to the requested type.

Unlike get, which returns Any, this method provides compile-time type safety. The stored value is returned as-is if it is already an instance of type_; otherwise the raw value is passed to type_(**raw) (for dict-like values) or type_(raw) as a fallback.

Parameters

Parameter	Type	Description
key	str	Cache key.
type_	type[_T]	The expected Python type of the cached value.
backend	str | None	Backend name (uses default if `None`).
default	_T | None	Value returned when the key is absent or deserialisation fails.

Returns

Type	Description
_T | None	The cached value cast to *type_*, or *default*.

---

CacheServiceConfig

source

Configuration for the cache service layer.

Controls service-level behavior and features like stampede protection, metrics, and circuit breaker.

Attributes: enable_protection: Enable cache stampede protection. enable_metrics: Enable service-level metrics. enable_health_checks: Enable health checks. protection_lock_ttl: Lock TTL for stampede protection. protection_max_wait: Max wait time for locks. protection_retry_interval: Retry interval for locks. default_backend: Name of the default backend. circuit_breaker_enabled: Enable circuit breaker pattern. circuit_breaker_threshold: Failure threshold for circuit breaker. default_serializer: Default serializer type.

---

CacheStats

source

to_dict

def to_dict() -> dict[str, Any]

source

---

CacheStatus

source

Status values for cache operations.

---

CachedRepositoryMixin

source

Mixin adding read-through caching and targeted invalidation to repositories.

Subclasses set: _cache: CacheBackendProtocol or CacheProviderProtocol instance cache_prefix: Namespace key prefix (e.g. "pet", "breed") cache_ttl: Entry time-to-live in seconds

cache

property cache() -> Any

source

Return the underlying cache backend/provider.

---

CompressingSerializer

source

AsyncStringSerializerProtocol with automatic compression for large values.

Compresses values above size threshold to save memory.

__init__

def __init__(
    compression_threshold: int = const.DEFAULT_COMPRESSION_THRESHOLD,
    compression_level: int = const.DEFAULT_COMPRESSION_LEVEL,
    allowed_classes: tuple[type, Ellipsis] | None = None
) -> None

source

Initialize serializer.

Parameters

Parameter	Type	Description
compression_threshold	int	Compress values larger than this (bytes)
compression_level	int	Compression level (1-9, higher = more compression)
allowed_classes	tuple[type, Ellipsis] | None	Classes that may be reconstructed from cache payloads. Defaults to `None` — deny every class (fail closed).

Note

Cache payloads may originate from an attacker-controlled store (e.g. a shared Redis). Reconstruction is deny-by-default: only explicitly allowlisted classes are ever resolved; plain builtin containers (dicts, lists, strings, numbers) always round-trip.

serialize

async def serialize(value: Any) -> str

source

Serialize and optionally compress value.

Parameters

Parameter	Type	Description
value	Any	Value to serialize

Returns

Type	Description
str	Serialized string (possibly compressed)

deserialize

async def deserialize(value: str) -> Any

source

Deserialize and decompress if needed.

Parameters

Parameter	Type	Description
value	str	Serialized string

Returns

Type	Description
Any	Original value

Raises

Exception	Description
SerializationError	If deserialization fails

get_stats

def get_stats() -> dict[str, Any]

source

Get serializer statistics.

Returns

Type	Description
dict[str, Any]	Stats dict

---

CostAwareCacheDecision

source

Cost-aware cache hit/miss decision function.

Implements a simple heuristic that balances accuracy (similarity score) against the economic cost of invoking the API. When API costs are high, lower similarity thresholds become acceptable to save money. When costs are low, higher similarity thresholds are enforced.

The decision is based on:

• similarity: Cosine similarity [0, 1]. Closeness to 1 = high accuracy.
• api_cost_per_1k_tokens: USD cost per 1000 tokens.
• expected_tokens: Expected token count for a fresh API call.

__init__

def __init__(accuracy_weight: float = 0.7) -> None

source

Initialize the cost-aware decision function.

Parameters

Parameter	Type	Description
accuracy_weight	float	Weight (0 to 1) assigned to accuracy mismatch penalty. Higher values prioritize accuracy over cost. Defaults to 0.7.

should_use_cache

def should_use_cache(
    similarity: float,
    api_cost_per_1k_tokens: float,
    expected_tokens: int
) -> bool

source

Decide whether to use a cached response.

Decision logic:

• Compute mismatch_penalty = (1 - similarity) * accuracy_weight
• Compute cost_incentive = min(estimated_cost * (1 - accuracy_weight), 1.0)
• Return mismatch_penalty < cost_incentive

When accuracy_weight=0.7 and similarity=0.95:

• mismatch_penalty = 0.05 * 0.7 = 0.035
• If estimated_cost >= 0.035/0.3 = 0.117 USD, cache is used.
• For gpt-4 (~$0.03 per 1k tokens), expected_tokens=100 → ~$0.003, cache is skipped (cost too low to accept small mismatch).
• For gpt-4 (~$0.03 per 1k tokens), expected_tokens=5000 → ~$0.15, cache is used (cost high enough to accept 5% mismatch).

Parameters

Parameter	Type	Description
similarity	float	Cosine similarity in [0, 1].
api_cost_per_1k_tokens	float	Estimated USD cost per 1000 tokens.
expected_tokens	int	Expected token count for fresh call.

Returns

Type	Description
bool	True if the cached response should be used, False if a fresh API call should be made.

---

DistributedLockInfo

source

Lock ownership information for distributed locks.

---

EnvironmentConfigLoader

source

Loader for cache configuration from various sources.

from_env

def from_env(prefix: str = const.ENV_PREFIX) -> CacheConfig

source

Load CacheConfig from environment variables.

Parameters

Parameter	Type	Description
prefix	str	Environment variable prefix.

Returns

Type	Description
CacheConfig	Populated CacheConfig instance.

from_dict

def from_dict(config_dict: dict[str, Any]) -> CacheConfig

source

Load CacheConfig from a dictionary.

Parameters

Parameter	Type	Description
config_dict	dict[str, Any]	Configuration dictionary.

Returns

Type	Description
CacheConfig	Populated CacheConfig instance.

from_yaml

def from_yaml(file_path: str) -> CacheConfig

source

Load CacheConfig from a YAML file.

Parameters

Parameter	Type	Description
file_path	str	Path to the YAML configuration file.

Returns

Type	Description
CacheConfig	Populated CacheConfig instance.

from_json

def from_json(file_path: str) -> CacheConfig

source

Load CacheConfig from a JSON file.

Parameters

Parameter	Type	Description
file_path	str	Path to the JSON configuration file.

Returns

Type	Description
CacheConfig	Populated CacheConfig instance.

---

FaissVectorIndex

source

FAISS-backed in-memory vector index for semantic cache.

Uses IndexFlatIP (inner product on L2-normalized vectors = cosine similarity). Suitable for single-server deployments with up to ~1M entries. Implements VectorIndexProtocol.

All vectors must be L2-normalized (unit length) before storage.

__init__

def __init__(
    embedding_dim: int = 384,
    max_entries: int = 100000
) -> None

source

Initialize the FAISS vector index.

Parameters

Parameter	Type	Description
embedding_dim	int	Dimension of embedding vectors. Defaults to 384 (common for sentence-transformers like all-MiniLM-L6-v2).
max_entries	int	Maximum number of entries before warning. Defaults to 100,000.

Raises

Exception	Description
ImportError	If faiss package is not installed.
ValueError	If embedding_dim <= 0 or max_entries <= 0.

search

async def search(
    embedding: list[float],
    k: int = 1
) -> list[tuple[str, float]]

source

Search for the top-k most similar embeddings.

Parameters

Parameter	Type	Description
embedding	list[float]	Query embedding vector (must be L2-normalized).
k	int	Number of results to return.

Returns

Type	Description
list[tuple[str, float]]	List of (cache_key, similarity_score) tuples, sorted by similarity_score descending. Empty list if index is empty.

add

async def add(
    key: str,
    embedding: list[float]
) -> None

source

Add an embedding to the index.

If the key already exists, it is not duplicated (early return). If the index reaches max_entries, a warning is logged but the entry is still added (LRU eviction can be a future enhancement).

Parameters

Parameter	Type	Description
key	str	Cache key identifier for the embedding.
embedding	list[float]	Embedding vector (must be L2-normalized).

remove

async def remove(key: str) -> bool

source

Remove an entry from the index by cache key.

FAISS IndexFlatIP does not support deletion. We use a soft-delete approach by marking the slot with None. Removed entries are skipped in search results.

Parameters

Parameter	Type	Description
key	str	Cache key identifier to remove.

Returns

Type	Description
bool	True if the key was found and removed, False if not found.

size

property size() -> int

source

Number of entries currently indexed (excluding deleted entries).

Returns

Type	Description
int	Non-negative integer.

---

HealthStatus

source

Unified health status.

---

InvalidatingCacheMixin

source

Mixin adding read-through caching and targeted invalidation to repositories.

Subclasses set: _cache: CacheBackendProtocol or CacheProviderProtocol instance cache_prefix: Namespace key prefix (e.g. "pet", "breed") cache_ttl: Entry time-to-live in seconds

cache

property cache() -> Any

source

Return the underlying cache backend/provider.

---

JSONSerializer

source

High-performance JSON-based serializer with enhanced type support.

Uses orjson (5-10x faster than stdlib json) for serialization with automatic fallback to stdlib json if orjson is unavailable.

Supports serialization of:

• Basic Python types (str, int, float, bool, None)
• Collections (list, dict, tuple, set)
• datetime and date objects (via orjson native support)
• UUID objects (via orjson native support)
• Custom objects with dict attribute

Performance:

• With orjson: ~5-10x faster than stdlib json
• Native support for datetime, UUID, dataclasses
• Optimized for Redis/Memcached serialization workloads

This class implements the AsyncStringSerializerProtocol protocol.

serialize

async def serialize(value: Any) -> str

source

Serialize a Python object to JSON string using orjson (5-10x faster).

deserialize

async def deserialize(value: str) -> Any

source

Deserialize a JSON string back to a Python object.

---

MemcachedBackendConfig

source

Configuration for Memcached cache backend.

Attributes: name: Backend name. default: Whether this is the default backend. enabled: Whether this backend is enabled. host: Memcached host. port: Memcached port. servers: List of server addresses (host:port). default_ttl: Default TTL in seconds. key_prefix: Key prefix for namespace isolation. connect_timeout: Connection timeout in seconds. timeout: Operation timeout in seconds. max_pool_size: Maximum connection pool size.

backend_type

property backend_type() -> BackendType

source

Get the backend type.

server_list

property server_list() -> list[str]

source

Get the server list.

---

MemcachedCacheBackend

source

Memcached cache backend implementation.

This backend provides Memcached-based distributed caching with TTL support. Unlike memory and Redis backends, this is implemented directly since the core infrastructure layer doesn’t provide a Memcached driver.

__init__

def __init__(
    servers: list[str],
    config: CacheOperationConfig | None = None
)

source

Initialize the Memcached cache backend.

Parameters

Parameter	Type	Description
servers	list[str]	List of Memcached server addresses (e.g., ["localhost:11211"]).
config	CacheOperationConfig | None	Cache configuration. If None, uses default configuration.

Raises

Exception	Description
ImportError	If pymemcache is not installed.

get

async def get(key: str) -> Any | None

source

Get a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to retrieve.

Returns

Type	Description
Any | None	The cached value if found and not expired, None otherwise.

set

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

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
key	str	The cache key to set.
value	Any	The value to cache.
ttl	int | None	Time to live in seconds. If None, uses default TTL.

Returns

Type	Description
bool	True if successful, False otherwise.

delete

async def delete(key: str) -> bool

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to delete.

Returns

Type	Description
bool	True if the key was deleted, False if it didn't exist or error occurred.

exists

async def exists(key: str) -> bool

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to check.

Returns

Type	Description
bool	True if the key exists and is not expired, False otherwise.

clear

async def clear() -> bool

source

Clear all values from the cache.

Note: This operation is not supported by Memcached protocol. Memcached doesn’t provide a way to clear all keys.

Returns

Type	Description
bool	False (operation not supported).

get_many

async def get_many(keys: list[str]) -> dict[str, Any]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to retrieve.

Returns

Type	Description
dict[str, Any]	Dictionary mapping keys to their values (missing keys omitted).

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> bool

source

Set multiple values in the cache.

Parameters

Parameter	Type	Description
items	dict[str, Any]	Dictionary of key-value pairs to cache.
ttl	int | None	Time to live in seconds for all items.

Returns

Type	Description
bool	True if all items were set successfully, False otherwise.

delete_many

async def delete_many(keys: list[str]) -> bool

source

Delete multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to delete.

Returns

Type	Description
bool	True if all keys were deleted successfully, False otherwise.

delete_pattern

async def delete_pattern(pattern: str) -> int

source

Delete keys matching a pattern.

Memcached does not natively support key enumeration or pattern matching. This method logs a warning and returns 0. If pattern-based invalidation is required, use Redis or the in-memory backend instead.

Parameters

Parameter	Type	Description
pattern	str	Glob pattern (not supported by Memcached).

Returns

Type	Description
int	Always 0 — Memcached does not support key enumeration.

get_stats

def get_stats() -> dict[str, int | float | str] | None

source

Return counters; Memcached does not expose portable key counts.

reset_stats

def reset_stats() -> None

source

Reset local counters without deleting Memcached values.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured health check result.

---

MemoryBackendConfig

source

Configuration for in-memory cache backend.

Attributes: name: Backend name. default: Whether this is the default backend. enabled: Whether this backend is enabled. default_ttl: Default TTL in seconds. max_size: Maximum number of entries. cleanup_interval: Cleanup interval in seconds. key_prefix: Key prefix for namespace isolation.

backend_type

property backend_type() -> BackendType

source

Get the backend type.

---

MemoryCacheBackend

source

Memory cache backend using quadkit-components MemoryStateStore.

This backend provides in-memory caching with TTL support by wrapping the standardized MemoryStateStore from quadkit-components.

__init__

def __init__(
    config: CacheOperationConfig | None = None,
    max_size: int | None = None,
    hooks: HookRegistryProtocol | None = None
) -> None

source

Initialize the memory cache backend.

Parameters

Parameter	Type	Description
config	CacheOperationConfig | None	Cache configuration. If None, uses default configuration.
max_size	int | None	Maximum number of cache entries. When the store is full the least-recently-used entry is evicted. `None` disables eviction (use only for tests / development).
hooks	HookRegistryProtocol | None	Optional hook registry for lifecycle emission.

get

async def get(key: str) -> Result[Any | None, CacheError]

source

Get a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to retrieve.

Returns

Type	Description
Result[Any | None, CacheError]	Ok(value) if found, Ok(None) if not found, Err(CacheError) on failure.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
key	str	The cache key to set.
value	Any	The value to cache.
ttl	int | None	Time to live in seconds. If None, uses default TTL.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

delete

async def delete(key: str) -> Result[bool, CacheError]

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to delete.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure.

exists

async def exists(key: str) -> Result[bool, CacheError]

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to check.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure.

clear

async def clear() -> Result[None, CacheError]

source

Clear all values from the cache.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

get_many

async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to retrieve.

Returns

Type	Description
Result[dict[str, Any], CacheError]	Ok(dict) mapping found keys to values, Err(CacheError) on failure.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set multiple values in the cache.

Parameters

Parameter	Type	Description
items	dict[str, Any]	Dictionary of key-value pairs to cache.
ttl	int | None	Time to live in seconds for all items.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if all items set successfully, Err(CacheError) on failure.

delete_many

async def delete_many(keys: list[str]) -> Result[int, CacheError]

source

Delete multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to delete.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

delete_pattern

async def delete_pattern(pattern: str) -> Result[int, CacheError]

source

Delete all keys matching a glob-style pattern.

Iterates over the in-memory store’s known keys (including the configured prefix) and deletes those that match pattern.

Parameters

Parameter	Type	Description
pattern	str	Glob pattern, e.g. `"pet:list:*"`.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

get_stats

def get_stats() -> dict[str, int | float | str] | None

source

Return backend statistics for the admin dashboard.

Returns

Type	Description
dict[str, int | float | str] | None	Dict with `hits, misses, evictions, and entries` counts, or None when statistics are unavailable.

reset_stats

def reset_stats() -> None

source

Reset counters without evicting cached entries.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured health check result.

---

RedisBackendConfig

source

Configuration for Redis cache backend.

Attributes: name: Backend name. default: Whether this is the default backend. enabled: Whether this backend is enabled. host: Redis host. port: Redis port. db: Redis database number. password: Redis password (optional). url: Full Redis URL (overrides host/port/db). default_ttl: Default TTL in seconds. key_prefix: Key prefix for namespace isolation. ssl: Whether to use SSL/TLS. connection_pool_size: Connection pool size. socket_timeout: Socket timeout in seconds. retry_on_timeout: Whether to retry on timeout.

backend_type

property backend_type() -> BackendType

source

Get the backend type.

redis_url

property redis_url() -> str

source

Get the full Redis URL.

---

RedisCacheBackend

source

Redis cache backend using StateStoreProtocol protocol.

This backend provides Redis-based distributed caching with TTL support by wrapping the standardized StateStoreProtocol protocol.

__init__

def __init__(
    store: StateStoreProtocol,
    config: CacheOperationConfig | None = None,
    hooks: HookRegistryProtocol | None = None
)

source

Initialize the Redis cache backend.

Parameters

Parameter	Type	Description
store	StateStoreProtocol	StateStoreProtocol implementation for key-value operations.
config	CacheOperationConfig | None	Cache configuration. If None, uses default configuration.
hooks	HookRegistryProtocol | None	Optional hook registry for lifecycle emission.

get_underlying_client

def get_underlying_client() -> Any | None

source

Return the underlying Redis client for app.state integration.

Returns the Redis client from the underlying StateStore, or None if not available (e.g., not a Redis backend).

get

async def get(key: str) -> Result[Any | None, CacheError]

source

Get a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to retrieve.

Returns

Type	Description
Result[Any | None, CacheError]	Ok(value) if found, Ok(None) if not found, Err(CacheError) on failure.

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set a value in the cache with optional TTL.

Parameters

Parameter	Type	Description
key	str	The cache key to set.
value	Any	The value to cache.
ttl	int | None	Time to live in seconds. If None, uses default TTL.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

delete

async def delete(key: str) -> Result[bool, CacheError]

source

Delete a value from the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to delete.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if deleted, Ok(False) if not found, Err(CacheError) on failure.

exists

async def exists(key: str) -> Result[bool, CacheError]

source

Check if a key exists in the cache.

Parameters

Parameter	Type	Description
key	str	The cache key to check.

Returns

Type	Description
Result[bool, CacheError]	Ok(True) if exists, Ok(False) otherwise, Err(CacheError) on failure.

clear

async def clear() -> Result[None, CacheError]

source

Clear all values from the cache.

Uses pattern-based deletion to clear only prefixed keys. This is safer than FLUSHDB as it only clears cache keys.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if successful, Err(CacheError) on failure.

get_many

async def get_many(keys: list[str]) -> Result[dict[str, Any], CacheError]

source

Get multiple values from the cache.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to retrieve.

Returns

Type	Description
Result[dict[str, Any], CacheError]	Ok(dict) mapping found keys to values, Err(CacheError) on failure.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> Result[None, CacheError]

source

Set multiple values in the cache using pipeline.

Parameters

Parameter	Type	Description
items	dict[str, Any]	Dictionary of key-value pairs to cache.
ttl	int | None	Time to live in seconds for all items.

Returns

Type	Description
Result[None, CacheError]	Ok(None) if all items set successfully, Err(CacheError) on failure.

delete_many

async def delete_many(keys: list[str]) -> Result[int, CacheError]

source

Delete multiple values from the cache using pipeline.

Parameters

Parameter	Type	Description
keys	list[str]	List of cache keys to delete.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

delete_pattern

async def delete_pattern(pattern: str) -> Result[int, CacheError]

source

Delete all Redis keys matching a glob-style pattern.

Uses SCAN (non-blocking) to find matching keys then DEL or pipeline to remove them. The configured key prefix is prepended to pattern automatically.

Parameters

Parameter	Type	Description
pattern	str	Glob pattern, e.g. `"pet:list:*"`.

Returns

Type	Description
Result[int, CacheError]	Ok(count) of deleted keys, Err(CacheError) on failure.

get_stats

def get_stats() -> dict[str, int | float | str] | None

source

Return backend statistics for the admin dashboard.

Returns

Type	Description
dict[str, int | float | str] | None	Dict with `hits, misses, evictions, and entries` counts, or None when statistics are unavailable.

reset_stats

def reset_stats() -> None

source

Reset counters without deleting Redis values.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Perform a health check on the cache backend.

Returns

Type	Description
HealthCheckResult	Structured health check result.

---

RedisDriver

source

Shared Redis connection logic

__init__

def __init__(
    url: str = '',
    prefix: str = '',
    client: Any | None = None
)

source

close

async def close() -> None

source

Close the Redis client, releasing its connection pool.

---

RedisLockStore

source

Redis implementation of LockStore with owner validation.

Locks are stored as the value of the key and contain a lock_id of the form <owner>:<uuid>. Release and extend operations are executed via Lua scripts that compare-and-delete or compare-and-expire atomically.

__init__

def __init__(
    url: str = '',
    prefix: str = '',
    client: Any | None = None
)

source

close

async def close() -> None

source

Close the Redis client, releasing its connection pool.

acquire

async def acquire(
    resource: str,
    ttl: int = 30,
    owner: str | EllipsisType = ...
) -> str | None

source

Acquire a lock.

Parameters

Parameter	Type	Description
resource	str	The resource to lock.
ttl	int	Lock TTL in seconds.
owner	str | EllipsisType	Owner identifier (required).

Returns

Type	Description
str | None	lock_id (str) on success, None on failure.

release

async def release(
    resource: str,
    lock_id: str
) -> None

source

Release a lock only if lock_id matches the current owner.

Raises

Exception	Description
LockNotHeldError	if the lock does not exist
LockOwnershipError	if the lock is held by someone else

extend

async def extend(
    resource: str,
    lock_id: str,
    ttl: int
) -> None

source

Extend TTL of a lock only if lock_id matches the current owner.

Raises

Exception	Description
LockNotHeldError	if the lock does not exist
LockOwnershipError	if the lock is held by someone else

is_locked

async def is_locked(key: str) -> bool

source

Check if a key is locked

locked

async def locked(
    resource: str,
    ttl: int = 30,
    owner: str | None = None
) -> Any

source

Async context manager that acquires a lock and releases it on exit.

Yields the lock_id string.

When owner is omitted a unique per-acquisition owner is generated (the lock is single-tenant: only this context can release it).

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check the health of the lock store

---

RedisSecretStore

source

Redis implementation of SecretStore with optional at-rest encryption.

Parameters

Parameter	Type	Description
url		Redis connection URL (e.g. `redis://localhost:6379/0`).
prefix		Optional string prefix applied to every key stored in Redis.
encryption_key		Optional Fernet symmetric encryption key (32-byte URL-safe base64-encoded bytes or string). When provided all secret values are encrypted before writing and decrypted on read. Generate one with:: import secrets, base64 key = base64.urlsafe_b64encode(secrets.token_bytes(32))

__init__

def __init__(
    url: str = '',
    prefix: str = '',
    encryption_key: bytes | str | None = None,
    client: Any | None = None
)

source

close

async def close() -> None

source

Close the Redis client, releasing its connection pool.

get_secret

async def get_secret(key: str) -> str | None

source

Get a secret by name, decrypting if encryption is configured.

set_secret

async def set_secret(
    key: str,
    value: str
) -> None

source

Set a secret, encrypting if encryption is configured.

delete_secret

async def delete_secret(key: str) -> None

source

Delete a secret

list_secrets

async def list_secrets(prefix: str | None = None) -> list[str]

source

List secrets with optional prefix

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check the health of the secret store

---

RedisStateStore

source

Redis implementation of StateStoreProtocol

get

async def get(key: str) -> Any | None

source

Get a value by key

set

async def set(
    key: str,
    value: Any,
    ttl: int | None = None
) -> None

source

Set a value with optional TTL

delete

async def delete(key: str) -> bool

source

Delete a value by key

exists

async def exists(key: str) -> bool

source

Check whether a key exists.

expire

async def expire(
    key: str,
    ttl: int
) -> bool

source

Set expiration on a key; False when the key does not exist.

ttl

async def ttl(key: str) -> int

source

Remaining TTL in seconds; -1 with no expiry, -2 when absent.

get_many

async def get_many(keys: list[str]) -> dict[str, Any]

source

Fetch several keys at once; absent keys are omitted.

set_many

async def set_many(
    items: dict[str, Any],
    ttl: int | None = None
) -> None

source

Persist several key-value pairs in one call.

get_bulk

async def get_bulk(keys: list[str]) -> dict[str, Any]

source

Get multiple values by keys

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check the health of the state store

---

SemanticCacheStore

source

Three-tier semantic cache: exact hash → vector similarity → miss.

Implements SemanticCacheProtocol. Caches LLM responses using a three-tier lookup strategy:

Tier 1 (Exact Hash): Normalizes and hashes the query to look for exact matches in the cache backend. Fast, deterministic.

Tier 2 (Vector Similarity): If Tier 1 misses, embeds the query and searches the vector index for similar cached queries. Returns the cached response if similarity >= similarity_threshold.

Tier 3 (Cache Miss): If both tiers miss, returns None for the caller to invoke the LLM.

The store() method populates both Tier 1 and Tier 2. The invalidate() method removes from both.

__init__

def __init__(
    cache_backend: CacheBackendProtocol,
    embedding_client: EmbeddingClientProtocol,
    vector_index: VectorIndexProtocol,
    similarity_threshold: float = 0.95,
    cache_ttl: int | None = None
) -> None

source

Initialize the semantic cache store.

Parameters

Parameter	Type	Description
cache_backend	CacheBackendProtocol	Backend for exact hash storage (e.g., Redis, in-memory cache). Implements CacheBackendProtocol.
embedding_client	EmbeddingClientProtocol	LLM embedding service. Implements EmbeddingClientProtocol.
vector_index	VectorIndexProtocol	Vector similarity search index. Implements VectorIndexProtocol.
similarity_threshold	float	Minimum cosine similarity to accept a Tier 2 cache hit. Must be in [0, 1]. Defaults to 0.95.
cache_ttl	int | None	Optional TTL in seconds for cache entries. Passed to cache_backend.set(). Defaults to None (no expiry).

Raises

Exception	Description
ValueError	If similarity_threshold is not in [0, 1].

lookup

async def lookup(query: str) -> str | None

source

Look up a query in the cache.

Checks Tier 1 (exact hash) then Tier 2 (vector similarity).

Parameters

Parameter	Type	Description
query	str	The user query string.

Returns

Type	Description
str | None	Cached response string, or None on cache miss.

store

async def store(
    query: str,
    response: str,
    model: str
) -> None

source

Store a query-response pair in both tiers.

Stores in Tier 1 (hash key) and Tier 2 (vector index).

Parameters

Parameter	Type	Description
query	str	The user query string.
response	str	The LLM response to cache.
model	str	The model that produced the response (for logging/tracking).

invalidate

async def invalidate(query: str) -> bool

source

Invalidate a cached entry by query.

Removes from both Tier 1 (cache backend) and Tier 2 (vector index).

Parameters

Parameter	Type	Description
query	str	The user query string to invalidate.

Returns

Type	Description
bool	True if the entry was found and removed, False otherwise.

---

TaggedCacheKey

source

---

TypeRegistry

source

Map ``(module, qualname)`` tags to model classes.

Registration validates that the class exposes a model_validate classmethod (fail fast at registration, not at gadget time).

Note

Cache data is untrusted input. An unregistered tag is never imported; it is refused by lookup.

__init__

def __init__() -> None

source

Initialize an empty registry.

register

def register(model_cls: type) -> None

source

Register a model class for reconstruction.

Parameters

Parameter	Type	Description
model_cls	type	Class exposing a `model_validate` classmethod.

Raises

Exception	Description
TypeError	If the class does not expose `model_validate`.

get

def get(
    module: str,
    qualname: str
) -> type[Any] | None

source

Return the registered class for a tag, or None when unregistered.

Parameters

Parameter	Type	Description
module	str	Module name of the type tag.
qualname	str	Qualified name of the type tag.

Returns

Type	Description
type[Any] | None	The registered class, or None.

clear

def clear() -> None

source

Remove all registrations (test isolation).

---

Functions

cache

cache

def cache(
    service: CacheService,
    key_prefix: str = 'cache',
    ttl: int | None = None,
    backend: str | None = None,
    protect: bool = True
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to cache function results with custom key generation.

Parameters

Parameter	Type	Description
service	CacheService	CacheService instance
key_prefix	str	Prefix for cache keys
ttl	int | None	Time-to-live in seconds
backend	str | None	Backend name
protect	bool	Enable stampede protection

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator function

Example

@cache(cache_service, key_prefix=“api”, ttl=300) async def get_user_data(user_id: int): return await fetch_from_database(user_id)

---

cache_evict

cache_evict

def cache_evict(
    *patterns: str,
    skip_on_error: bool = True
) -> Callable[[F], F]

source

Invalidate matching cache keys after a successful write method call.

Resolves self._cache (CacheBackendProtocol) on the target service instance and calls delete_pattern(pattern) for each glob pattern in patterns. When the wrapped method returns a Result in the Err state, eviction is skipped so failed writes do not thrash warm cache entries.

Parameters

Parameter	Type	Description
skip_on_error	bool	When `True`, backend eviction errors are logged and suppressed instead of failing the caller's successful write.

---

cache_in_request

cache_in_request

def cache_in_request(func: _F) -> _F

source

Cache the result of an async function for the duration of the current request.

The cache key is derived from the function’s qualified name and all positional and keyword arguments. Arguments must support a stable repr() so that logically equal calls produce the same key.

• Same request, same args → cached value returned; wrapped coroutine is not awaited again.
• Same request, different args → each distinct arg combination gets its own cache entry.
• Different requests → each request has an isolated contextvars.ContextVar context; they never share entries.

Parameters

Parameter	Type	Description
func	_F	The async callable to wrap. Must be a coroutine function.

Returns

Type	Description
_F	A coroutine function with identical signature that transparently caches results within the current request context.

Example

@cache_in_request
async def get_user_permissions(user_id: str) -> list[str]:
    return await db.fetch_permissions(user_id)

@cache_in_request
async def get_user_permissions(user_id: str) -> list[str]:
    return await db.fetch_permissions(user_id)

---

cacheable

cacheable

def cacheable(
    ttl: int | None = 300,
    key_prefix: str = '',
    skip_on_error: bool = True,
    *,
    models: Sequence[type[Any]] = ()
) -> Callable[[F], F]

source

Cache the return value of an async method using the injected CacheBackend.

The decorator resolves the cache backend from self._cache on the target instance. The cache key is derived from the method name, prefix, and a SHA-256 hash of the serialized arguments.

Supports methods that return Result[T, E]:

• Ok(value) results are cached; the inner value is unwrapped, serialised, and re-wrapped in Ok on retrieval.
• Err(...) results are never cached — errors propagate on every call.

Supports domain model return values (or Result-wrapped domain models):

• Domain models are serialised to a type-tagged JSON envelope and reconstructed via model_validate on retrieval, preserving type identity across the JSON round-trip.
• Reconstruction is deny-by-default: type tags are only resolved against the registered type registry (:data:~quadkit.cache.serialization.type_registry.DEFAULT_REGISTRY); a payload naming an unregistered type degrades to its raw data.
• Passing models=(ModelCls, ...) statically registers those domain model classes in DEFAULT_REGISTRY at decoration time.

Parameters

Parameter	Type	Description
ttl	int | None	Time-to-live in seconds. `None` disables expiry.
key_prefix	str	Optional prefix prepended to the generated cache key.
skip_on_error	bool	When `True, cache misses on backend errors gracefully fall through to the original method. When False`, errors propagate.
models	Sequence[type[Any]]	Optional sequence of domain model classes to register in `DEFAULT_REGISTRY` for safe reconstruction on cache hits.

Returns

Type	Description
Callable[[F], F]	Decorator that wraps an async method with cache-aside logic.

Example

class UserService:
    def __init__(self, cache: CacheBackendProtocol) -> None:
        self._cache = cache

    @cacheable(ttl=60, key_prefix="users", models=(User,))
    async def get_user(self, user_id: str) -> Result[User, DomainError]:
        return await self._repo.find(user_id)

class UserService:
    def __init__(self, cache: CacheBackendProtocol) -> None:
        self._cache = cache

    @cacheable(ttl=60, key_prefix="users", models=(User,))
    async def get_user(self, user_id: str) -> Result[User, DomainError]:
        return await self._repo.find(user_id)

---

clear_request_cache

clear_request_cache

def clear_request_cache() -> None

source

Reset the current request's cache to an empty state.

Call this at the end of a request (e.g. in ASGI middleware teardown) if you need an explicit eviction guarantee rather than relying on the context being discarded by the ASGI server.

---

conditional_cache

conditional_cache

def conditional_cache(
    service: CacheService,
    condition_func: Callable[[Any], bool],
    key_prefix: str = 'cond',
    ttl: int | None = None,
    backend: str | None = None,
    protect: bool = True
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to cache function results only when condition is met.

Parameters

Parameter	Type	Description
service	CacheService	CacheService instance
condition_func	Callable[[Any], bool]	Function that returns True if result should be cached
key_prefix	str	Prefix for cache keys
ttl	int | None	Time-to-live in seconds
backend	str | None	Backend name
protect	bool	Enable stampede protection

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator function

Example

@conditional_cache( cache_service, lambda result: result is not None, ttl=300 ) async def search_database(query: str): return await db.search(query)

---

get_request_cache

get_request_cache

def get_request_cache() -> dict[str, Any]

source

Return the current request's cache dict, creating it if needed.

The first call within a new request context allocates a fresh dict and stores it via ContextVar.set so that all subsequent calls in the same context share it.

Returns

Type	Description
dict[str, Any]	The mutable cache dict for this request context.

---

invalidate_cache

invalidate_cache

def invalidate_cache(
    service: CacheService,
    key_pattern: str,
    backend: str | None = None
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to invalidate cache keys after function execution.

Parameters

Parameter	Type	Description
service	CacheService	CacheService instance
key_pattern	str	Pattern for keys to invalidate (supports wildcards)
backend	str | None	Backend name

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator function

Example

@invalidate_cache(cache_service, “user:*”) async def update_user(user_id: int, data: dict): return await db.update_user(user_id, data)

---

remember

remember

def remember(
    service: CacheService,
    ttl: int | None = None,
    backend: str | None = None,
    protect: bool = True
) -> Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]

source

Decorator to remember function results (memoization).

Uses automatic key generation based on function signature.

Parameters

Parameter	Type	Description
service	CacheService	CacheService instance
ttl	int | None	Time-to-live in seconds
backend	str | None	Backend name
protect	bool	Enable stampede protection

Returns

Type	Description
Callable[[Callable[Ellipsis, Any]], Callable[Ellipsis, Any]]	Decorator function

Example

@remember(cache_service, ttl=600) async def expensive_computation(x: int, y: int): return await slow_calculation(x, y)

---

Exceptions

CacheBackendError

source

Raised when the cache backend (Redis/Memcached) fails.

---

CacheCapacityError

source

Raised when the cache is at capacity.

---

CacheConfigurationError

source

Raised when cache configuration is invalid.

__init__

def __init__(
    message: str = const.ERROR_MSG_CACHE_CONFIGURATION,
    setting: str | None = None,
    value: Any | None = None,
    **kwargs: Any
)

source

---

CacheConnectionError

source

Raised when the cache connection fails.

---

CacheError

source

Base exception for cache errors.

---

CacheInvalidationError

source

Raised when cache invalidation fails.

__init__

def __init__(
    message: str = const.ERROR_MSG_CACHE_INVALIDATION,
    keys: list[str] | None = None,
    tag: str | None = None,
    pattern: str | None = None,
    **kwargs: Any
)

source

---

CacheKeyError

source

Raised when a cache key is invalid or for key-specific errors.

---

CacheSerializationError

source

Raised when serialization/deserialization fails.

---

CacheStampedeError

source

Raised when cache stampede protection fails.

__init__

def __init__(
    message: str = const.ERROR_MSG_CACHE_STAMPEDE,
    lock_holder: str | None = None,
    **kwargs: Any
)

source

---

CacheTimeoutError

source

Raised when a cache operation times out.

__init__

def __init__(
    message: str = const.ERROR_MSG_CACHE_TIMEOUT,
    key: str | None = None,
    backend: str | None = None,
    timeout_seconds: float | None = None,
    **kwargs: Any
)

source

---