Cache Utilities¶

Public API for the caching layer, covering the decorator entry point and the strategies that back it. The module keeps a consistent contract across sync and async callables, so you can cache coroutines and legacy helpers without branching the implementation.

Tip

Pair this reference with Cache for the feature guide.

Quick overview¶

cache wraps any callable and auto-detects async functions so you do not need two decorators.
Strategy selection follows the standard priority chain: keyword arguments > kstlib.conf.yml > presets.
TTL, LRU, and file-backed strategies ship out of the box; custom strategies can extend CacheStrategy.
Every wrapped function exposes cache_clear() and cache_info() helpers for test hygiene and observability.

Configuration cascade¶

The decorator consults the loaded config each time it needs to spawn (or respawn) a strategy. A minimal config block looks like:

cache:
    default_strategy: ttl
    ttl:
        default_seconds: 600
        max_entries: 2048

You can override any of these knobs per call site:

from kstlib.cache import cache

@cache(strategy="file", cache_dir="/tmp/kst-cache", serializer="auto")
def expensive_lookup(key: str) -> dict:
    return fetch_remote_payload(key)

Decorator¶

cache¶

kstlib.cache.cache(func: F) → F[source]

Cache decorator with automatic async/sync detection.

Supports multiple caching strategies configured via kstlib.conf.yml or decorator arguments. Automatically detects async functions and handles them appropriately.

Parameters:

func – Function to cache (when used without parentheses)
strategy – Cache strategy (‘ttl’, ‘lru’, ‘file’)
ttl – Time to live in seconds (TTL strategy)
maxsize – Maximum cache size (LRU strategy)
cache_dir – Cache directory path (file strategy)
check_mtime – Check file modification time (file strategy)
serializer – Serialization format for file strategy (‘json’, ‘pickle’, ‘auto’)

Returns:

Decorated function with caching

Examples

Basic usage (uses config defaults):

>>> @cache
... def double(x: int) -> int:
...     return x * 2
>>> double(5)
10
>>> double(5)  # Returns cached value
10

With explicit TTL strategy:

>>> @cache(strategy="ttl", ttl=60)
... def compute(n: int) -> int:
...     return n * n
>>> compute(4)
16

LRU cache for recursive functions:

>>> @cache(strategy="lru", maxsize=128)
... def fibonacci(n: int) -> int:
...     if n < 2:
...         return n
...     return fibonacci(n - 1) + fibonacci(n - 2)
>>> fibonacci(10)
55

Cache management methods:

>>> double.cache_info()
{'strategy': 'ttl', 'is_async': False}
>>> double.cache_clear()  # Clear cached values

Strategies¶

CacheStrategy¶

class kstlib.cache.CacheStrategy[source]

Bases: ABC

Abstract base class for cache strategies.

All cache strategies must implement get() and set() methods to store and retrieve cached values.

abstract get(self, key: 'str') → 'Any | None' -> Any | None[source]

Retrieve value from cache.

Parameters:: key (str) – Cache key
Returns:: Cached value or None if not found/expired
Return type:: Any | None

abstract set(self, key: 'str', value: 'Any') → 'None' -> None[source]

Store value in cache.

Parameters:

key (str) – Cache key
value (Any) – Value to cache

abstract clear(self) → 'None' -> None[source]

Clear all cached values.

static make_key(func: 'Callable[..., Any]', args: 'tuple[Any, ...]', kwargs: 'dict[str, Any]') → 'str' -> str[source]

Generate cache key from function and arguments.

Normalizes function calls by binding arguments to signature, ensuring that process(1, 2) and process(1, 2, c=0) produce the same cache key when c has default value 0.

Parameters:

func (Callable[[...], Any]) – Function being cached
args (tuple[Any, ...]) – Positional arguments
kwargs (dict[str, Any]) – Keyword arguments

Returns:

Hash-based cache key

Return type:

str

TTLCacheStrategy¶

class kstlib.cache.TTLCacheStrategy(ttl=300, max_entries=1000, cleanup_interval=60)[source]

Bases: CacheStrategy

Time-To-Live cache strategy.

Caches values with expiration time. Expired entries are removed automatically during cleanup or access.

Parameters:

ttl (int) – Time to live in seconds
max_entries (int) – Maximum number of cache entries
cleanup_interval (int) – Seconds between cleanup runs

Examples

>>> cache = TTLCacheStrategy(ttl=300, max_entries=1000)
>>> cache.set("key1", "value1")
>>> cache.get("key1")
'value1'

__init__(self, ttl: 'int' = 300, max_entries: 'int' = 1000, cleanup_interval: 'int' = 60) → 'None' -> None[source]

Initialize TTL cache strategy.

get(self, key: 'str') → 'Any | None' -> Any | None[source]

Retrieve value from cache if not expired.

Parameters:: key (str) – Cache key
Returns:: Cached value or None if expired/not found
Return type:: Any | None

set(self, key: 'str', value: 'Any') → 'None' -> None[source]

Store value in cache with TTL.

Parameters:

key (str) – Cache key
value (Any) – Value to cache

clear(self) → 'None' -> None[source]

Clear all cached values.

LRUCacheStrategy¶

class kstlib.cache.LRUCacheStrategy(maxsize=128, typed=False)[source]

Bases: CacheStrategy

Least Recently Used cache strategy.

Wraps functools.lru_cache for compatibility with CacheStrategy interface.

Parameters:

maxsize (int) – Maximum cache size
typed (bool) – If True, cache different types separately

Examples

>>> cache = LRUCacheStrategy(maxsize=128)
>>> cache.set("key1", "value1")
>>> cache.get("key1")
'value1'

__init__(self, maxsize: 'int' = 128, typed: 'bool' = False) → 'None' -> None[source]

Initialize LRU cache strategy.

get(self, key: 'str') → 'Any | None' -> Any | None[source]

Retrieve value from cache and update access order.

Parameters:: key (str) – Cache key
Returns:: Cached value or None if not found
Return type:: Any | None

set(self, key: 'str', value: 'Any') → 'None' -> None[source]

Store value in cache with LRU eviction.

Parameters:

key (str) – Cache key
value (Any) – Value to cache

clear(self) → 'None' -> None[source]

Clear all cached values.

FileCacheStrategy¶

class kstlib.cache.FileCacheStrategy(cache_dir='.cache', check_mtime=True, serializer='json', memory_max_entries=256, limits=None)[source]

Bases: CacheStrategy

File-based cache with mtime checking.

Caches function results based on file modification time and persists values on disk using JSON serialization by default. A pickle-based fallback can be enabled explicitly for trusted environments or automatically by using the "auto" serializer.

Parameters:

cache_dir (str) – Directory for cache files.
check_mtime (bool) – If True, invalidate cache on file modification.
serializer (str) – Serialization format ("json" | "pickle" | "auto").
memory_max_entries (int | None) – Max entries to retain in memory cache.
limits (CacheLimits | None) – Optional CacheLimits for config-driven size limits.

Examples

>>> cache = FileCacheStrategy(cache_dir=".cache", check_mtime=True)
>>> # Cache will invalidate if the source file changes

DEFAULT_MEMORY_MAX_ENTRIES = 256: Default maximum entries for in-memory cache layer.

__init__(self, cache_dir: 'str' = '.cache', check_mtime: 'bool' = True, serializer: 'str' = 'json', memory_max_entries: 'int | None' = 256, limits: 'CacheLimits | None' = None) → 'None' -> None[source]

Initialize file cache strategy.

get(self, key: 'str') → 'Any | None' -> Any | None[source]

Retrieve value from cache.

Parameters:: key (str) – Cache key
Returns:: Cached value or None if not found/invalid
Return type:: Any | None

set(self, key: 'str', value: 'Any', source_path: 'Path | None' = None) → 'None' -> None[source]

Store value in cache.

Parameters:

key (str) – Cache key
value (Any) – Value to cache
source_path (Path | None) – Optional source file path for mtime tracking

clear(self) → 'None' -> None[source]

Clear all cached values.

Configuration Limits¶

CacheLimits¶

class kstlib.limits.CacheLimits(max_file_size)[source]

Bases: object

Resolved cache resource limits.

max_file_size: int

property max_file_size_display: str: Human-readable cache file size limit.

__init__(self, max_file_size: 'int') → None -> None

get_cache_limits¶

kstlib.limits.get_cache_limits(config: 'Mapping[str, Any] | None' = None) → 'CacheLimits' -> CacheLimits[source]

Resolve cache limits from config with hard limit enforcement.

Parameters:: config (Mapping[str, Any] | None) – Optional config mapping. If None, loads from get_config().
Returns:: CacheLimits with resolved values clamped to hard maximums.
Return type:: CacheLimits