API Reference

Public API

YokedCache - A robust, performance-focused caching library for Python backends.

YokedCache provides seamless caching integration for FastAPI applications with Redis, featuring automatic cache invalidation, fuzzy search capabilities, and intelligent database integration.

CacheBackend

Bases: ABC

Abstract base class for cache backends.

is_connected: bool property

Check if backend is connected.

connect() -> None abstractmethod async

Establish connection to the backend.

delete(key: str) -> bool abstractmethod async

Delete key from cache.

disconnect() -> None abstractmethod async

Close connection to the backend.

exists(key: str) -> bool abstractmethod async

Check if key exists in cache.

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

Set expiration time for existing key.

flush_all() -> bool abstractmethod async

Flush all cache keys.

fuzzy_search(query: str, threshold: int = 80, max_results: int = 10, tags: Optional[Set[str]] = None) -> List[FuzzySearchResult] abstractmethod async

Perform fuzzy search on cached data.

get(key: str, default: Any = None) -> Any abstractmethod async

Get value from cache.

get_all_keys(pattern: str = '*') -> List[str] abstractmethod async

Get all keys matching pattern.

get_size_bytes() -> int abstractmethod async

Get total size of cache in bytes.

get_stats() -> CacheStats abstractmethod async

Get current cache statistics.

health_check() -> bool abstractmethod async

Check if the backend connection is healthy.

invalidate_pattern(pattern: str) -> int abstractmethod async

Invalidate all keys matching a pattern.

invalidate_tags(tags: Union[str, List[str], Set[str]]) -> int abstractmethod async

Invalidate all keys associated with given tags.

set(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Set[str]] = None) -> bool abstractmethod async

Set value in cache.

CacheConnectionError

Bases: YokedCacheError

Raised when Redis connection fails or times out.

CacheEntry dataclass

Represents a single cache entry with metadata.

age_seconds: float property

Get the age of the cache entry in seconds.

is_expired: bool property

Check if the cache entry has expired.

touch() -> None

Update the last accessed timestamp and increment hit count.

CacheKeyError

Bases: YokedCacheError

Raised when cache key operations fail.

CacheMetrics

Cache metrics wrapper that handles multiple collectors.

add_collector(collector: MetricsCollector) -> None

Add a metrics collector.

end_timer(timer_id: str, operation: str) -> None async

End timing an operation and record the duration.

gauge(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Set gauge metric across all collectors.

histogram(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record histogram across all collectors.

increment(metric: str, value: int = 1, tags: Optional[Dict[str, str]] = None) -> None async

Increment metric across all collectors.

start_timer(operation: str) -> str

Start timing an operation.

timing(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record timing across all collectors.

CacheSerializationError

Bases: YokedCacheError

Raised when data serialization/deserialization fails.

CacheStats dataclass

Cache performance and usage statistics.

hit_rate: float property

Calculate cache hit rate as a percentage.

memory_usage: int property

Alias for total_memory_bytes for backward compatibility.

miss_rate: float property

Calculate cache miss rate as a percentage.

add_hit(table: Optional[str] = None, tags: Optional[Set[str]] = None) -> None

Record a cache hit.

add_miss(table: Optional[str] = None, tags: Optional[Set[str]] = None) -> None

Record a cache miss.

InvalidationRule dataclass

Defines when and how to invalidate cache entries.

should_invalidate(operation_type: InvalidationType) -> bool

Check if this rule should trigger for the given operation type.

MemcachedBackend

Bases: CacheBackend

Memcached cache backend implementation.

connect() -> None async

Establish connection to Memcached.

delete(key: str) -> bool async

Delete key from cache.

disconnect() -> None async

Close Memcached connection.

exists(key: str) -> bool async

Check if key exists in cache.

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

Set expiration time for existing key.

flush_all() -> bool async

Flush all cache keys.

fuzzy_search(query: str, threshold: int = 80, max_results: int = 10, tags: Optional[Set[str]] = None) -> List[FuzzySearchResult] async

Perform fuzzy search on cached data.

get(key: str, default: Any = None) -> Any async

Get value from cache.

get_all_keys(pattern: str = '*') -> List[str] async

Get all keys matching pattern.

get_size_bytes() -> int async

Get total size of cache in bytes.

get_stats() -> CacheStats async

Get current cache statistics.

health_check() -> bool async

Check if Memcached connection is healthy.

invalidate_pattern(pattern: str) -> int async

Invalidate all keys matching a pattern.

invalidate_tags(tags: Union[str, List[str], Set[str]]) -> int async

Invalidate all keys associated with given tags.

set(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Set[str]] = None) -> bool async

Set value in cache.

MemoryBackend

Bases: CacheBackend

In-memory cache backend implementation.

connect() -> None async

Establish connection (start cleanup task).

delete(key: str) -> bool async

Delete key from cache.

disconnect() -> None async

Close connection (stop cleanup task).

exists(key: str) -> bool async

Check if key exists in cache.

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

Set expiration time for existing key.

flush_all() -> bool async

Flush all cache keys with the configured prefix.

fuzzy_search(query: str, threshold: int = 80, max_results: int = 10, tags: Optional[Set[str]] = None) -> List[FuzzySearchResult] async

Perform fuzzy search on cached data.

get(key: str, default: Any = None) -> Any async

Get value from cache.

get_all_keys(pattern: str = '*') -> List[str] async

Get all keys matching pattern.

get_size_bytes() -> int async

Get total size of cache in bytes.

get_stats() -> CacheStats async

Get current cache statistics.

health_check() -> bool async

Check if memory backend is healthy.

invalidate_pattern(pattern: str) -> int async

Invalidate all keys matching a pattern.

invalidate_tags(tags: Union[str, List[str], Set[str]]) -> int async

Invalidate all keys associated with given tags.

set(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Set[str]] = None) -> bool async

Set value in cache.

NoOpCollector

Bases: MetricsCollector

No-op metrics collector for when monitoring is disabled.

gauge(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

No-op gauge.

histogram(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

No-op histogram.

increment(metric: str, value: int = 1, tags: Optional[Dict[str, str]] = None) -> None async

No-op increment.

timing(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

No-op timing.

PrometheusCollector

Bases: MetricsCollector

Prometheus metrics collector.

gauge(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Set a gauge metric.

histogram(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record a histogram value.

increment(metric: str, value: int = 1, tags: Optional[Dict[str, str]] = None) -> None async

Increment a counter metric.

timing(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record a timing metric.

RedisBackend

Bases: CacheBackend

Redis cache backend implementation.

connect() -> None async

Establish connection to Redis.

delete(key: str) -> bool async

Delete key from cache.

disconnect() -> None async

Close Redis connection.

exists(key: str) -> bool async

Check if key exists in cache.

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

Set expiration time for existing key.

flush_all() -> bool async

Flush all cache keys with the configured prefix.

fuzzy_search(query: str, threshold: int = 80, max_results: int = 10, tags: Optional[Set[str]] = None) -> List[FuzzySearchResult] async

Perform fuzzy search on cached data.

get(key: str, default: Any = None) -> Any async

Get value from cache.

get_all_keys(pattern: str = '*') -> List[str] async

Get all keys matching pattern.

get_size_bytes() -> int async

Get total size of cache in bytes.

get_stats() -> CacheStats async

Get current cache statistics.

health_check() -> bool async

Check if Redis connection is healthy.

invalidate_pattern(pattern: str) -> int async

Invalidate all keys matching a pattern.

invalidate_tags(tags: Union[str, List[str], Set[str]]) -> int async

Invalidate all keys associated with given tags.

set(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Set[str]] = None) -> bool async

Set value in cache.

RedisVectorSearch

Redis-specific vector search with persistence.

delete_vector(key: str) -> bool async

Delete a vector from Redis.

get_vector(key: str) async

Retrieve a vector from Redis.

store_vector(key: str, vector) -> bool async

Store a vector in Redis.

StatsDCollector

Bases: MetricsCollector

StatsD metrics collector.

gauge(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Set a gauge metric.

histogram(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record a histogram value.

increment(metric: str, value: int = 1, tags: Optional[Dict[str, str]] = None) -> None async

Increment a counter metric.

timing(metric: str, value: Union[int, float], tags: Optional[Dict[str, str]] = None) -> None async

Record a timing metric.

VectorSimilaritySearch

Vector-based similarity search for cache entries.

fit(cache_data: Dict[str, Any]) -> None

Fit the vectorizer on cache data.

Parameters:

Name	Type	Description	Default
cache_data	Dict[str, Any]	Dictionary of cache key -> value pairs	required

get_stats() -> Dict[str, Any]

Get statistics about the vector search index.

remove_cache_entry(key: str) -> None

Remove a cache entry from the search index.

Parameters:

Name	Type	Description	Default
key	str	Cache key to remove	required

search(query: str, cache_data: Dict[str, Any], threshold: float = 0.1, max_results: int = 10) -> List[FuzzySearchResult]

Perform vector-based similarity search.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
cache_data	Dict[str, Any]	Dictionary of cache key -> value pairs	required
threshold	float	Similarity threshold (0.0 - 1.0)	0.1
max_results	int	Maximum number of results	10

Returns:

Type	Description
List[FuzzySearchResult]	List of fuzzy search results sorted by similarity score

update_cache_entry(key: str, value: Any) -> None

Update a single cache entry in the search index.

Parameters:

Name	Type	Description	Default
key	str	Cache key	required
value	Any	Cache value	required

YokedCacheError

Bases: Exception

Base exception class for all YokedCache errors.

cached(cache: Optional[YokedCache] = None, ttl: Optional[int] = None, key_prefix: Optional[str] = None, tags: Optional[Union[str, List[str], Set[str]]] = None, table: Optional[str] = None, serialization: Optional[SerializationMethod] = None, skip_cache_on_error: bool = True, key_builder: Optional[Callable[..., str]] = None, key_func: Optional[Callable[..., str]] = None, condition: Optional[Callable[..., bool]] = None) -> Callable[[F], F]

Decorator to cache function results.

Parameters:

Name	Type	Description	Default
cache	Optional[YokedCache]	YokedCache instance (if None, creates default)	None
ttl	Optional[int]	Cache TTL in seconds	None
key_prefix	Optional[str]	Custom key prefix for this function	None
tags	Optional[Union[str, List[str], Set[str]]]	Tags for cache invalidation	None
table	Optional[str]	Database table name for auto-invalidation	None
serialization	Optional[SerializationMethod]	Serialization method	None

Returns:

Type	Description
Callable[[F], F]	Decorated function

cached_dependency(cache_or_func: Optional[Union[YokedCache, Callable[..., Any]]] = None, *, cache: Optional[YokedCache] = None, ttl: Optional[int] = None, key_prefix: Optional[str] = None, table_name: Optional[str] = None, auto_invalidate: bool = True, dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = None) -> Union[Callable[..., Any], Callable[[Callable[..., Any]], Callable[..., Any]]]

cached_dependency(cache_or_func: Callable[..., Any], *, cache: Optional[YokedCache] = ..., ttl: Optional[int] = ..., key_prefix: Optional[str] = ..., table_name: Optional[str] = ..., auto_invalidate: bool = ..., dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = ...) -> Callable[..., Any]

cached_dependency(cache_or_func: Optional[YokedCache] = ..., *, cache: Optional[YokedCache] = ..., ttl: Optional[int] = ..., key_prefix: Optional[str] = ..., table_name: Optional[str] = ..., auto_invalidate: bool = ..., dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = ...) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Wrap a FastAPI dependency with caching.

This is specifically designed for database dependencies like get_db(). Properly handles both regular functions and generator functions.

Can be used in two ways: 1. As decorator: @cached_dependency(cache, dependencies=["user:123"]) 2. As function: cached_dependency(func, cache=cache)

Parameters:

Name	Type	Description	Default
cache_or_func	Optional[Union[YokedCache, Callable[..., Any]]]	Either YokedCache instance or function to wrap	None
cache	Optional[YokedCache]	YokedCache instance (when using function style)	None
ttl	Optional[int]	Cache TTL in seconds	None
key_prefix	Optional[str]	Custom key prefix	None
table_name	Optional[str]	Table name for auto-invalidation	None
auto_invalidate	bool	Enable auto-invalidation on writes	True
dependencies	Optional[Union[str, List[str], Callable[..., Iterable[str]]]]	Cache dependencies for invalidation	None

Decorator function that wraps the dependency function or wrapped

deserialize_data(data: bytes, method: SerializationMethod = SerializationMethod.JSON) -> Any

Deserialize data using the specified method.

Parameters:

Name	Type	Description	Default
data	bytes	Serialized data as bytes	required
method	SerializationMethod	Serialization method that was used	JSON

Returns:

Type	Description
Any	Deserialized data

Raises:

Type	Description
CacheSerializationError	If deserialization fails

generate_cache_key(prefix: str, table: Optional[str] = None, query: Optional[str] = None, params: Optional[Dict[str, Any]] = None, user_id: Optional[Union[str, int]] = None, namespace: Optional[str] = None) -> str

Generate a standardized cache key.

Parameters:

Name	Type	Description	Default
prefix	str	Cache key prefix (usually app name)	required
table	Optional[str]	Database table name	None
query	Optional[str]	SQL query or operation identifier	None
params	Optional[Dict[str, Any]]	Query parameters or filters	None
user_id	Optional[Union[str, int]]	User identifier for user-specific caching	None
namespace	Optional[str]	Additional namespace for multi-tenancy	None

Returns:

Type	Description
str	Formatted cache key string

serialize_data(data: Any, method: SerializationMethod = SerializationMethod.JSON) -> bytes

Serialize data using the specified method.

Parameters:

Name	Type	Description	Default
data	Any	Data to serialize	required
method	SerializationMethod	Serialization method to use	JSON

Returns:

Type	Description
bytes	Serialized data as bytes

Raises:

Type	Description
CacheSerializationError	If serialization fails

Core

Cache

Core YokedCache implementation.

This module contains the main YokedCache class that provides the primary caching functionality, including Redis integration, auto-invalidation, and cache management operations.

YokedCache

Main caching class that provides intelligent caching with Redis backend.

Features: - Automatic cache invalidation based on database operations - Variable TTLs per table/query type - Tag-based cache grouping and invalidation - Fuzzy search capabilities - Performance metrics and monitoring - Async/await support for FastAPI integration

adelete(*args, **kwargs) -> bool async

Explicitly async version of delete.

aexists(*args, **kwargs) -> bool async

Explicitly async version of exists.

aget(*args, **kwargs) -> Any async

Explicitly async version of get.

aset(*args, **kwargs) -> bool async

Explicitly async version of set.

close() -> None async

Close the cache connection.

connect() -> None async

Establish connection to Redis.

Tests patch redis.asyncio.from_url; use that path for compatibility.

delete(key: str) -> bool async

Delete key from cache.

Parameters:

Name	Type	Description	Default
key	str	Cache key to delete	required

Returns:

Type	Description
bool	True if key was deleted, False if key didn't exist

delete_sync(key: str) -> bool

Sync version of delete().

detailed_health_check() -> Dict[str, Any] async

Comprehensive health check for monitoring.

Returns detailed information about cache health, performance, and system status suitable for monitoring dashboards.

disconnect() -> None async

Close Redis connection.

exists(key: str) -> bool async

Check if key exists in cache.

exists_sync(key: str) -> bool

Sync version of exists().

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

Set expiration time for existing key.

flush() -> bool async

Flush all cache data.

flush_all() -> bool async

Flush all cache keys with the configured prefix.

Returns:

Type	Description
bool	True if successful

fuzzy_search(query: str, threshold: int = 80, max_results: int = 10, tags: Optional[Set[str]] = None) -> List[FuzzySearchResult] async

Perform fuzzy search on cached data.

Parameters:

Name	Type	Description	Default
query	str	Search query	required
threshold	int	Similarity threshold (0-100)	80
max_results	int	Maximum number of results	10
tags	Optional[Set[str]]	Optional tags to filter by	None

Returns:

Type	Description
List[FuzzySearchResult]	List of fuzzy search results

get(key: str, default: Any = None, touch: bool = True) -> Any async

Get value from cache.

Parameters:

Name	Type	Description	Default
key	str	Cache key	required
default	Any	Default value if key not found	None
touch	bool	Whether to update access time and hit count	True

Returns:

Type	Description
Any	Cached value or default

get_comprehensive_metrics() -> Dict[str, Any] async

Get comprehensive metrics including enhanced performance data.

get_stats() -> CacheStats async

Get current cache statistics.

get_sync(key: str, default: Any = None, touch: bool = True) -> Any

Sync version of get() with proper async context detection.

Warns when used in async context and suggests using aget() instead.

health() -> bool async

Check if the cache is healthy.

health_check() -> bool async

Check if Redis connection is healthy.

invalidate_by_tags(tags: Union[str, List[str], Set[str]]) -> int async

Alias for invalidate_tags for backward compatibility.

invalidate_pattern(pattern: str) -> int async

Invalidate all keys matching a pattern.

Parameters:

Name	Type	Description	Default
pattern	str	Redis pattern (supports * and ? wildcards)	required

Returns:

Type	Description
int	Number of keys invalidated

invalidate_tags(tags: Union[str, List[str], Set[str]]) -> int async

Invalidate all keys associated with given tags.

Parameters:

Name	Type	Description	Default
tags	Union[str, List[str], Set[str]]	Tags to invalidate	required

Returns:

Type	Description
int	Number of keys invalidated

ping() -> bool async

Ping the cache backend.

set(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Union[str, List[str], Set[str]]] = None, serialization: Optional[SerializationMethod] = None) -> bool async

Set value in cache.

Parameters:

Name	Type	Description	Default
key	str	Cache key	required
value	Any	Value to cache	required
ttl	Optional[int]	Time to live in seconds	None
tags	Optional[Union[str, List[str], Set[str]]]	Tags for grouping and invalidation	None
serialization	Optional[SerializationMethod]	Serialization method to use	None

Returns:

Type	Description
bool	True if successful, False otherwise

set_sync(key: str, value: Any, ttl: Optional[int] = None, tags: Optional[Union[str, List[str], Set[str]]] = None, serialization: Optional[SerializationMethod] = None) -> bool

Sync version of set().

start_metrics_collection() -> None

Start background metrics collection if enabled.

stop_metrics_collection() -> None async

Stop background metrics collection.

Decorators

Decorators for YokedCache integration.

This module provides decorators for easy integration with FastAPI and other Python frameworks, enabling automatic caching of functions and dependencies.

CachedDatabaseWrapper

Wrapper for database sessions that adds caching capabilities.

This wrapper intercepts database queries and automatically caches results while also handling cache invalidation on write operations.

cache property

Access to the cache instance.

pending_invalidations property

Get pending write operations that will trigger invalidation.

session property

Access to the underlying database session.

commit() async

Handle commit operations with cache invalidation.

invalidate_pending() async

Invalidate pending cache entries based on write operations.

cached(cache: Optional[YokedCache] = None, ttl: Optional[int] = None, key_prefix: Optional[str] = None, tags: Optional[Union[str, List[str], Set[str]]] = None, table: Optional[str] = None, serialization: Optional[SerializationMethod] = None, skip_cache_on_error: bool = True, key_builder: Optional[Callable[..., str]] = None, key_func: Optional[Callable[..., str]] = None, condition: Optional[Callable[..., bool]] = None) -> Callable[[F], F]

Decorator to cache function results.

Parameters:

Name	Type	Description	Default
cache	Optional[YokedCache]	YokedCache instance (if None, creates default)	None
ttl	Optional[int]	Cache TTL in seconds	None
key_prefix	Optional[str]	Custom key prefix for this function	None
tags	Optional[Union[str, List[str], Set[str]]]	Tags for cache invalidation	None
table	Optional[str]	Database table name for auto-invalidation	None
serialization	Optional[SerializationMethod]	Serialization method	None

Returns:

Type	Description
Callable[[F], F]	Decorated function

cached_dependency(cache_or_func: Optional[Union[YokedCache, Callable[..., Any]]] = None, *, cache: Optional[YokedCache] = None, ttl: Optional[int] = None, key_prefix: Optional[str] = None, table_name: Optional[str] = None, auto_invalidate: bool = True, dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = None) -> Union[Callable[..., Any], Callable[[Callable[..., Any]], Callable[..., Any]]]

cached_dependency(cache_or_func: Callable[..., Any], *, cache: Optional[YokedCache] = ..., ttl: Optional[int] = ..., key_prefix: Optional[str] = ..., table_name: Optional[str] = ..., auto_invalidate: bool = ..., dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = ...) -> Callable[..., Any]

cached_dependency(cache_or_func: Optional[YokedCache] = ..., *, cache: Optional[YokedCache] = ..., ttl: Optional[int] = ..., key_prefix: Optional[str] = ..., table_name: Optional[str] = ..., auto_invalidate: bool = ..., dependencies: Optional[Union[str, List[str], Callable[..., Iterable[str]]]] = ...) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Wrap a FastAPI dependency with caching.

This is specifically designed for database dependencies like get_db(). Properly handles both regular functions and generator functions.

Can be used in two ways: 1. As decorator: @cached_dependency(cache, dependencies=["user:123"]) 2. As function: cached_dependency(func, cache=cache)

Parameters:

Name	Type	Description	Default
cache_or_func	Optional[Union[YokedCache, Callable[..., Any]]]	Either YokedCache instance or function to wrap	None
cache	Optional[YokedCache]	YokedCache instance (when using function style)	None
ttl	Optional[int]	Cache TTL in seconds	None
key_prefix	Optional[str]	Custom key prefix	None
table_name	Optional[str]	Table name for auto-invalidation	None
auto_invalidate	bool	Enable auto-invalidation on writes	True
dependencies	Optional[Union[str, List[str], Callable[..., Iterable[str]]]]	Cache dependencies for invalidation	None

Decorator function that wraps the dependency function or wrapped

warm_cache(cache: YokedCache, functions_to_warm: List[Dict[str, Any]]) -> int async

Warm cache by pre-executing functions.

Parameters:

Name	Type	Description	Default
cache	YokedCache	YokedCache instance	required
functions_to_warm	List[Dict[str, Any]]	List of function configurations [{"func": function, "args": [], "kwargs": {}, "ttl": 300}, ...]	required

Returns:

Type	Description
int	Number of functions successfully warmed

Configuration

Configuration management for YokedCache.

This module handles loading and managing configuration from files, environment variables, and programmatic settings.

CacheConfig dataclass

Main configuration class for YokedCache.

add_table_config(config: TableCacheConfig) -> None

Add or update table-specific configuration.

get_connection_pool_config() -> Dict[str, Any]

Get connection pool configuration including custom kwargs.

get_table_config(table_name: str) -> TableCacheConfig

Get configuration for a specific table, with fallbacks.

to_dict() -> Dict[str, Any]

Convert configuration to dictionary.

create_default_config() -> CacheConfig

Create a default configuration instance.

load_config_from_file(file_path: Union[str, Path]) -> CacheConfig

Load configuration from a YAML file.

save_config_to_file(config: CacheConfig, file_path: Union[str, Path]) -> None

Save configuration to a YAML file.

Models

Data models and structures for YokedCache.

This module defines the core data structures used throughout YokedCache, including cache entries, statistics, and configuration models.

CacheEntry dataclass

Represents a single cache entry with metadata.

age_seconds: float property

Get the age of the cache entry in seconds.

is_expired: bool property

Check if the cache entry has expired.

touch() -> None

Update the last accessed timestamp and increment hit count.

CacheOperation dataclass

Represents a cache operation for logging/debugging.

CacheStats dataclass

Cache performance and usage statistics.

hit_rate: float property

Calculate cache hit rate as a percentage.

memory_usage: int property

Alias for total_memory_bytes for backward compatibility.

miss_rate: float property

Calculate cache miss rate as a percentage.

add_hit(table: Optional[str] = None, tags: Optional[Set[str]] = None) -> None

Record a cache hit.

add_miss(table: Optional[str] = None, tags: Optional[Set[str]] = None) -> None

Record a cache miss.

ConnectionPoolStats dataclass

Redis connection pool statistics.

FuzzySearchResult dataclass

Result from a fuzzy search operation.

InvalidationRule dataclass

Defines when and how to invalidate cache entries.

should_invalidate(operation_type: InvalidationType) -> bool

Check if this rule should trigger for the given operation type.

InvalidationType

Bases: Enum

Types of cache invalidation triggers.

SerializationMethod

Bases: Enum

Supported data serialization methods.

TableCacheConfig dataclass

Cache configuration for a specific database table.

Utils

Utility functions for YokedCache.

This module provides common utility functions for key generation, data serialization, hashing, and other helper operations.

calculate_ttl_with_jitter(base_ttl: int, jitter_percent: float = 10.0) -> int

Calculate TTL with random jitter to prevent thundering herd.

Parameters:

Name	Type	Description	Default
base_ttl	int	Base TTL in seconds	required
jitter_percent	float	Percentage of jitter to add (0-100)	10.0

Returns:

Type	Description
int	TTL with jitter applied

deserialize_data(data: bytes, method: SerializationMethod = SerializationMethod.JSON) -> Any

Deserialize data using the specified method.

Parameters:

Name	Type	Description	Default
data	bytes	Serialized data as bytes	required
method	SerializationMethod	Serialization method that was used	JSON

Returns:

Type	Description
Any	Deserialized data

Raises:

Type	Description
CacheSerializationError	If deserialization fails

extract_table_from_query(query: str) -> Optional[str]

Extract table name from SQL query (simple heuristic).

Parameters:

Name	Type	Description	Default
query	str	SQL query string	required

Returns:

Type	Description
Optional[str]	Extracted table name or None if not found

format_bytes(bytes_value: int) -> str

Format bytes value in human-readable format.

Parameters:

Name	Type	Description	Default
bytes_value	int	Number of bytes	required

Returns:

Type	Description
str	Formatted string (e.g., "1.5 MB")

generate_cache_key(prefix: str, table: Optional[str] = None, query: Optional[str] = None, params: Optional[Dict[str, Any]] = None, user_id: Optional[Union[str, int]] = None, namespace: Optional[str] = None) -> str

Generate a standardized cache key.

Parameters:

Name	Type	Description	Default
prefix	str	Cache key prefix (usually app name)	required
table	Optional[str]	Database table name	None
query	Optional[str]	SQL query or operation identifier	None
params	Optional[Dict[str, Any]]	Query parameters or filters	None
user_id	Optional[Union[str, int]]	User identifier for user-specific caching	None
namespace	Optional[str]	Additional namespace for multi-tenancy	None

Returns:

Type	Description
str	Formatted cache key string

get_current_timestamp() -> float

Get current UTC timestamp.

get_operation_type_from_query(query: str) -> str

Determine operation type from SQL query.

Parameters:

Name	Type	Description	Default
query	str	SQL query string	required

Returns:

Type	Description
str	Operation type ('select', 'insert', 'update', 'delete', 'unknown')

normalize_tags(tags: Union[str, List[str], Set[str]]) -> Set[str]

Normalize tags to a consistent set format.

Parameters:

Name	Type	Description	Default
tags	Union[str, List[str], Set[str]]	Tags in various formats	required

Returns:

Type	Description
Set[str]	Normalized set of tags

parse_redis_url(url: str) -> Dict[str, Any]

Parse Redis URL into connection parameters.

Parameters:

Name	Type	Description	Default
url	str	Redis URL (e.g., redis://user:pass@host:port/db)	required

Returns:

Type	Description
Dict[str, Any]	Dictionary of connection parameters

sanitize_key(key: str) -> str

Sanitize cache key to ensure Redis compatibility.

Parameters:

Name	Type	Description	Default
key	str	Original cache key	required

Returns:

Type	Description
str	Sanitized cache key

serialize_data(data: Any, method: SerializationMethod = SerializationMethod.JSON) -> bytes

Serialize data using the specified method.

Parameters:

Name	Type	Description	Default
data	Any	Data to serialize	required
method	SerializationMethod	Serialization method to use	JSON

Returns:

Type	Description
bytes	Serialized data as bytes

Raises:

Type	Description
CacheSerializationError	If serialization fails

timestamp_to_datetime(timestamp: float) -> datetime

Convert timestamp to UTC datetime.

timing_decorator(func)

Decorator to measure function execution time.

timing_decorator_async(func)

Async decorator to measure function execution time.