ahvn.cache.base module¶

Abstract cache protocol and common helpers.

Implements the CacheEntry structure and BaseCache with memoization decorators for sync/async functions and streaming generators.

class ahvn.cache.base.CacheEntry(func, inputs=<factory>, output=Ellipsis, expected=Ellipsis, metadata=<factory>, _key=None)[source]¶

Bases: object

A universal object to store cache entries, containing the function (name), inputs, output, and optional metadata.

Parameters:

func (str)
inputs (Dict[str, Any])
output (Any)
expected (Any)
metadata (Dict[str, Any] | None)
_key (int)

func: str¶

inputs: Dict[str, Any]¶

output: Any = Ellipsis¶

expected: Any = Ellipsis¶

metadata: Dict[str, Any] | None¶

property key¶

property value¶

classmethod from_args(func='', output=..., expected=..., metadata=None, exclude=None, **inputs)[source]¶

Creates a CacheEntry from function arguments.

Parameters:

func (Union[Callable, str]) – The function or its name to be cached.
output (Any) – The output of the function.
expected (Any) – The expected output of the function.
metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry
exclude (Optional[Iterable[str]]) – Keys to exclude from inputs.
**inputs (Any) – Arbitrary keyword arguments representing the inputs to the function.

Return type:

CacheEntry

classmethod from_dict(data, exclude=None)[source]¶

Creates a CacheEntry from a dictionary.

Parameters:

data (Dict[str, Any]) – The dictionary containing the cache entry data.
exclude (Optional[Iterable[str]]) – Keys to exclude from inputs.

Return type:

CacheEntry

to_dict()[source]¶

Converts the CacheEntry to a dictionary.

Returns:: A dictionary representation of the CacheEntry.
Return type:: Dict[str, Any]

clone(**updates)[source]¶

Creates a clone of the CacheEntry with optional updates to its attributes.

Parameters:: **updates – Arbitrary keyword arguments to update the CacheEntry attributes.
Returns:: A new CacheEntry instance with updated attributes.
Return type:: CacheEntry

annotate(expected=..., metadata=None)[source]¶

Annotates the CacheEntry with expected output and metadata.

Parameters:

expected (Any) – The expected output of the function. If omitted, will use the actual output as annotation.
metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.

Returns:

A new CacheEntry instance with the annotation.

Return type:

CacheEntry

property annotated: bool¶

Checks if the CacheEntry has been annotated with expected output.

Returns:: True if the CacheEntry has an expected output, False otherwise.
Return type:: bool

__init__(func, inputs=<factory>, output=Ellipsis, expected=Ellipsis, metadata=<factory>, _key=None)¶

Parameters:

func (str)
inputs (Dict[str, Any])
output (Any)
expected (Any)
metadata (Dict[str, Any] | None)
_key (int)

Return type:

None

class ahvn.cache.base.BaseCache(exclude=None, *args, **kwargs)[source]¶

Bases: ABC

An abstract base class for cache implementations.

The class provides memoize and batch_memoize decorators to cache function results. The class requires subclasses to implement storage and retrieval methods. The caching supports both synchronous and asynchronous functions.

Parameters:: exclude (Iterable[str] | None)

_cache¶

A dictionary to store cache entries.

Type:: Dict[str, CacheEntry]

Abstract Methods:: _get(key, default): Retrieves a dict (CacheEntry) from the cache by its key. Use Ellipsis as default to raise KeyError if not found (to avoid collisions with functions returning None). _set(key, value): Sets a cache entry in the cache. _remove(key): Removes a cache entry from the cache by its key. _itervalues(): Returns an iterator over the values in the cache. _clear(): Clears the cache.
Optional Override Methods:: _has(key): Determines if a cache entry exists for the given key. __len__(): Returns the number of entries in the cache.

Notes

The default implementations of __len__, _has, and batch operations are not optimized and may be slow for large stores. Subclasses are encouraged to override these for efficiency.
It is worth noticing that only the __getitem__ (implemented by _get) and add (implemented by _set) are used during memoize.

__init__(exclude=None, *args, **kwargs)[source]¶

Initialization.

Parameters:

exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.
*args – Additional positional arguments.
**kwargs – Additional keyword arguments.

set_exclude(exclude=None)[source]¶

Set the keys to exclude from inputs when creating cache entries.

Parameters:: exclude (Optional[Iterable[str]]) – Keys to exclude from inputs when creating cache entries.

add_exclude(exclude=None)[source]¶

Add keys to exclude from inputs when creating cache entries.

Parameters:: exclude (Optional[Iterable[str]]) – Keys to add to the exclusion list.

__contains__(key)[source]¶

Checks if a cache entry exists for the given key.

Parameters:: key (Union[int, CacheEntry]) – The key or CacheEntry to check in the cache.
Returns:: True if the cache entry exists, False otherwise.
Return type:: bool

exists(func, **kwargs)[source]¶

Checks if a cache entry exists for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to check in the cache.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

True if the cache entry exists, False otherwise.

Return type:

bool

__getitem__(key)[source]¶

Retrieves a cache entry for the given function and inputs.

Parameters:: key (Union[int, str, CacheEntry]) – The key or CacheEntry to retrieve.
Returns:: The cached entry if found. Otherwise Ellipsis.
Return type:: CacheEntry

retrieve(func, **kwargs)[source]¶

Retrieves a cached entry for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to retrieve the cached value for.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

The cached output if found, otherwise Ellipsis (to avoid collisions with functions returning None).

Return type:

Any

get(func, **kwargs)[source]¶

Gets a cached value for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to retrieve the cached value for.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

Returns:

The cached output if found, otherwise Ellipsis (to avoid collisions with functions returning None).

Return type:

Any

__setitem__(key, value)[source]¶

Sets a cache entry for the given function and inputs.

Parameters:

key (Union[int, str, CacheEntry]) – The key or CacheEntry to set.
value (Union[Dict[str, Any], CacheEntry]) – The value to cache.

set(func, output=..., expected=..., metadata=None, **kwargs)[source]¶

Sets a cached value for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to cache the value for.
output (Any) – The output to cache.
expected (Any) – The expected output of the function.
metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

add(entry)[source]¶

Sets a cache entry by directly adding it to the cache.

Parameters:: entry (CacheEntry) – The cache entry to add.

annotate(func, expected=..., metadata=None, **kwargs)[source]¶

Annotates a cached value for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to annotate the cached value for.
expected (Any) – The expected output of the function. If the cache entry exists, its expected will be modified, but its output will remain unchanged. Otherwise, a new entry with output and expected both set as the annotation will be created. If set to … and the cache entry exists, it will use the actual output as the annotation. If set to … but the cache entry does not exist, raise an error.
metadata (Optional[Dict[str, Any]]) – Optional metadata to store with the cache entry.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

Raises:

ValueError – If the original cache entry does not exist and expected is not provided.

__delitem__(key)[source]¶

Deletes a cache entry for the given function and inputs.

Parameters:: key (Union[int, str, CacheEntry]) – The key or CacheEntry of the cache entry to delete.

unset(func, **kwargs)[source]¶

Deletes a cache entry for the given function and inputs.

Parameters:

func (Union[Callable, str]) – The function or its name to delete the cache entry for.
**kwargs – Arbitrary keyword arguments representing the inputs to the function.

remove(entry)[source]¶

Deletes a cache entry by directly removing it from the cache.

Parameters:: entry (CacheEntry) – The cache entry to remove.

__iter__()[source]¶

Iterates over the cache entries.

Yields:: CacheEntry – The CacheEntry objects in the cache.
Return type:: Generator[CacheEntry, None, None]

pop()[source]¶

Pops an arbitrary cache entry from the cache.

Returns:: The popped CacheEntry if the cache is not empty, otherwise None.
Return type:: Optional[CacheEntry]

popall()[source]¶

Pops all cache entries from the cache.

Returns:: A list of all popped CacheEntry objects.
Return type:: List[CacheEntry]

clear()[source]¶: Clears the cache.

close()[source]¶: Closes the cache, if applicable.

flush(**kwargs)[source]¶: Flushes the cache, if applicable.

memoize(func=None, *, name=None)[source]¶

Decorator (or decorator factory) to cache the output of a function based on its inputs.

Return type:

Callable

Parameters:

func (Callable | None)
name (str | None)

Usage:

@cache.memoize def f(…): …

@cache.memoize(name=”xxx”) def g(…): …

When name is provided, it is used as the function identifier in the cache key. Otherwise, the wrapped function’s name is used.

batch_memoize(func=None, *, name=None)[source]¶

Decorator (or decorator factory) to cache the output of a function based on its inputs in batch mode.

Usage:

@cache.batch_memoize def f(…): …

@cache.batch_memoize(name=”xxx”) def g(…): …

Parameters:

func (Callable) – The function to cache.
name (Optional[str]) – Optional name for the function in the cache key.

Returns:

The wrapped function with batch caching.

Return type:

Callable