fastcfg.cache package

Submodules

fastcfg.cache.policies module

This module defines pre-configured cache policies for use with the caching mechanism.

fastcfg.cache.policies.TEN_MIN_TTL

A TTL (Time-To-Live) cache strategy with a

Type:: TTLCacheStrategy

duration of 10 minutes.

fastcfg.cache.policies.ONE_HOUR_TTL

A TTL (Time-To-Live) cache strategy with a duration of 1 hour.

Type:: TTLCacheStrategy

fastcfg.cache.policies.DAILY_TTL

A TTL (Time-To-Live) cache strategy with a duration of 1 day.

Type:: TTLCacheStrategy

fastcfg.cache.policies.LRU_POLICY

A Least Recently Used (LRU) cache strategy

Type:: LRUCacheStrategy

with a default capacity of 100 entries.

fastcfg.cache.policies.MRU_POLICY

A Most Recently Used (MRU) cache strategy

Type:: MRUCacheStrategy

with a default capacity of 100 entries.

fastcfg.cache.store module

This module provides a global store for managing multiple cache instances.

Classes:

CacheStore: A class that provides a centralized store for cache instances, allowing for adding, retrieving, and clearing caches globally.

Global Variables:

cache_store (CacheStore): A global instance of the CacheStore class for managing cache instances.

Usage Example:

from fastcfg.cache.store import cache_store from fastcfg.cache import Cache

# Create a new cache instance # Automatically added to global store internally my_cache = Cache(name=’my_cache’)

# Retrieve the cache from the global store retrieved_cache = cache_store.get_cache(‘my_cache’)

# Clear a specific cache cache_store.clear_cache(‘my_cache’)

# Clear all caches cache_store.clear_all_caches()

class fastcfg.cache.store.CacheStore[source]

Bases: object

A global store for managing multiple cache instances.

Purpose:

This class provides a centralized store for cache instances.
It allows for adding, retrieving, and clearing caches globally.

_caches

A set to store cache instances.

Type:: set

add_cache(cache)[source]: Adds a cache instance to the global store.

clear_all_caches()[source]: Clears all caches in the global store.

clear_cache(cache_name)[source]: Clears a specific cache by name.

get_cache(cache_name)[source]: Retrieves a specific cache by name.

add_cache(cache: None)[source]

Adds a cache to the global store.

Parameters:

cache_name (str) – The name of the cache.
cache (Cache) – The cache instance to add.

Raises:

ValueError – If a cache with the same name already exists.

clear_all_caches()[source]: Clears all caches globally.

clear_cache(cache_name: str)[source]

Clears a specific cache.

Parameters:: cache_name (str) – The name of the cache to clear.

get_cache(cache_name: str) → None[source]

Retrieves a specific cache if it exists.

Parameters:: cache_name (str) – The name of the cache to retrieve.
Returns:: The cache instance if found, otherwise None.
Return type:: Optional[Cache]

fastcfg.cache.strategies module

This module defines various cache strategies for use with the caching mechanism.

Classes:

TTLCacheStrategy (ICacheStrategy): A cache strategy that invalidates entries based on a time-to-live (TTL) value.

Attributes:
- _seconds (int): The TTL value in seconds.
Methods:
- __init__(seconds: int): Initialize the strategy with a TTL value.
- is_valid(meta_value: Optional[float]) -> bool: Check if the cache entry is still valid based on the TTL.
- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Set the TTL for a cache entry upon insertion.
- on_invalidation(key: str, cache: Cache) -> None: Perform any invalidation cleanup for a given cache key.

LRUCacheStrategy (IUsageCacheStrategy): A cache strategy that evicts the least recently used (LRU) entries when capacity is exceeded.

Methods:
- on_access(key: str, meta: Dict[str, Any]) -> None: Update the access order to mark the key as most recently used.
- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Handle insertion and evict if necessary.

MRUCacheStrategy (IUsageCacheStrategy): A cache strategy that evicts the most recently used (MRU) entries when capacity is exceeded.

Methods:
- on_access(key: str, meta: Dict[str, Any]) -> None: Update the access order to mark the key as most recently used.
- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Handle insertion and evict if necessary.

class fastcfg.cache.strategies.LRUCacheStrategy(capacity: int)[source]

Bases: AbstractUsageCacheStrategy

A cache strategy that evicts the least recently used (LRU) entries when capacity is exceeded.

- on_access(key: str, meta: Dict[str, Any]) -> None: Update the access order to mark the key as most recently used.

- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Handle insertion and evict if necessary.

on_access(key: str, meta: Dict[str, Any]) → None[source]

Update the access order to mark the key as most recently used.

Parameters:

key (str) – The cache key.
meta (Dict[str, Any]) – The metadata dictionary.

on_insertion(key: str, value: Any, meta: Dict[str, Any]) → None[source]

Handle insertion and evict if necessary.

Parameters:

key (str) – The cache key.
value (Any) – The cache value.
meta (Dict[str, Any]) – The metadata dictionary.

class fastcfg.cache.strategies.MRUCacheStrategy(capacity: int)[source]

Bases: AbstractUsageCacheStrategy

A cache strategy that evicts the most recently used (MRU) entries when capacity is exceeded.

- on_access(key: str, meta: Dict[str, Any]) -> None: Update the access order to mark the key as most recently used.

- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Handle insertion and evict if necessary.

on_access(key: str, meta: Dict[str, Any]) → None[source]

Update the access order to mark the key as most recently used.

Parameters:

key (str) – The cache key.
meta (Dict[str, Any]) – The metadata dictionary.

on_insertion(key: str, value: Any, meta: Dict[str, Any]) → None[source]

Handle insertion and evict if necessary.

Parameters:

key (str) – The cache key.
value (Any) – The cache value.
meta (Dict[str, Any]) – The metadata dictionary.

class fastcfg.cache.strategies.TTLCacheStrategy(seconds: int)[source]

Bases: AbstractCacheStrategy

A cache strategy that invalidates entries based on a time-to-live (TTL) value.

_seconds

The TTL value in seconds.

Type:: int

- __init__(seconds: int): Initialize the strategy with a TTL value.

- is_valid(meta_value: Optional[float]) -> bool: Check if the cache entry is still valid based on the TTL.

- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Set the TTL for a cache entry upon insertion.

- on_invalidation(key: str, cache: Cache) -> None: Perform any invalidation cleanup for a given cache key.

is_valid(meta_value: float | None) → bool[source]

Check if the cache entry is still valid based on the TTL.

Parameters:: meta_value (Optional[float]) – The metadata value to check.
Returns:: True if the entry is still valid, False otherwise.
Return type:: bool

on_insertion(key: str, value: Any, meta: Dict[str, Any]) → None[source]

Set the TTL for a cache entry upon insertion.

Parameters:

key (str) – The cache key.
value (Any) – The cache value.
meta (Dict[str, Any]) – The metadata dictionary.

on_invalidation(key: str, cache: Cache) → None[source]

Perform any invalidation cleanup for a given cache key.

Parameters:

key (str) – The cache key.
cache (Cache) – The cache instance.

Module contents

This module provides the core classes and interfaces for implementing cache strategies in the fastcfg library.

Classes:

AbstractCacheStrategy: An abstract base class defining the interface for cache strategies.
AbstractUsageCacheStrategy: An abstract base class for usage-based cache eviction strategies.
Cache: A class that manages cache entries using a specified cache strategy.

Exceptions:

MissingCacheKeyError: Raised when a requested cache key is not found.

Modules:

cache_store: Provides a global store for managing cache instances.

class fastcfg.cache.AbstractCacheStrategy[source]

Bases: ABC

An abstract base class that defines the interface for cache strategies.

- is_valid(meta_value: Any) -> bool: Determine if a cache entry is still valid.

- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Execute policy upon cache insertion.

- on_invalidation(key: str, cache: ‘Cache’) -> Optional[Any]: Performs any invalidation cleanup for a given cache key and optionally returns a default value.

- on_access(key: str, meta: Dict[str, Any]) -> None: Update metadata or perform actions upon cache access.

abstractmethod is_valid(meta_value: Any) → bool[source]: Determine if the cache entry is still valid based on the strategy.

on_access(key: str, meta: Dict[str, Any]) → None[source]: Update metadata or perform actions upon cache access.

abstractmethod on_insertion(key: str, value: Any, meta: Dict[str, Any]) → None[source]: Execute cache strategy policy upon insertion.

on_invalidation(key: str, cache: Cache) → Any | None[source]: Performs any invalidation cleanup for a given cache key and optionally returns a default value.

class fastcfg.cache.AbstractUsageCacheStrategy(capacity: int)[source]

Bases: AbstractCacheStrategy, ABC

An abstract base class of ICacheStrategy that implements usage-based cache eviction.

- __init__(capacity: int): Initialize the strategy with a given capacity.

- is_valid(meta_value: Optional[Any]) -> bool: Determine if a cache entry is valid based on usage.

- _remove_excess_entries(meta: Dict[str, Any], to_remove_key: str) -> None: Remove excess entries if capacity is exceeded.

- on_invalidation(key: str, cache: ‘Cache’) -> None: Perform any invalidation cleanup for a given cache key.

- on_insertion(key: str, value: Any, meta: Dict[str, Any]) -> None: Execute cache strategy policy upon insertion.

- on_access(key: str, meta: Dict[str, Any]) -> None: Update metadata or perform actions upon cache access.

is_valid(meta_value: Any | None) → bool[source]

Determine if a cache entry is valid based on usage.

Parameters:: meta_value (Optional[Any]) – The metadata value to check.
Returns:: True if the entry is valid, False otherwise.
Return type:: bool

abstractmethod on_access(key: str, meta: Dict[str, Any]) → None[source]: Update metadata or perform actions upon cache access.

abstractmethod on_insertion(key: str, value: Any, meta: Dict[str, Any]) → None[source]: Execute cache strategy policy upon insertion.

on_invalidation(key: str, cache: Cache) → None[source]

Perform any invalidation cleanup for a given cache key.

Parameters:

key (str) – The key of the entry to invalidate.
cache (Cache) – The cache instance.

class fastcfg.cache.Cache(cache_strategy: AbstractCacheStrategy, name: str = None)[source]

Bases: object

A class that manages cache entries using a specified cache strategy.

- __init__(cache_strategy: ICacheStrategy): Initialize the cache with a given strategy.

- set_value(key: str, value: Any) -> None: Set the value and associated metadata for a given key.

- get_value(key: str) -> Any: Retrieve the value for a given key if it’s valid.

- is_valid(key: str) -> bool: Check if a key is present and valid in the cache.

- get_metadata(key: str) -> Optional[Any]: Get metadata associated with a given cache key.

get_metadata(key: str) → Any | None[source]: Get metadata associated with a given cache key.

get_value(key: str) → Any[source]: Retrieve the value for a given key from the cache if it’s valid.

is_valid(key: str) → bool[source]: Check if a key is present and valid in the cache.

set_value(key: str, value: Any) → None[source]: Set the value and associated metadata for a given key in the cache.