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)

基类：object

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

参数:

• 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)

Creates a CacheEntry from function arguments.

参数:

• 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.

返回类型:

CacheEntry

classmethod from_dict(data, exclude=None)

Creates a CacheEntry from a dictionary.

参数:

• data (Dict[str, Any]) -- The dictionary containing the cache entry data.

• exclude (Optional[Iterable[str]]) -- Keys to exclude from inputs.

返回类型:

CacheEntry

to_dict()

Converts the CacheEntry to a dictionary.

返回:

A dictionary representation of the CacheEntry.

返回类型:

Dict[str, Any]

clone(**updates)

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

参数:

**updates -- Arbitrary keyword arguments to update the CacheEntry attributes.

返回:

A new CacheEntry instance with updated attributes.

返回类型:

CacheEntry

annotate(expected=..., metadata=None)

Annotates the CacheEntry with expected output and metadata.

参数:

• 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.

返回:

A new CacheEntry instance with the annotation.

返回类型:

CacheEntry

property annotated: bool

Checks if the CacheEntry has been annotated with expected output.

返回:

True if the CacheEntry has an expected output, False otherwise.

返回类型:

bool

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

参数:

• func (str)

• inputs (Dict[str, Any])

• output (Any)

• expected (Any)

• metadata (Dict[str, Any] | None)

• _key (int)

返回类型:

None

class ahvn.cache.base.BaseCache(exclude=None, *args, **kwargs)

基类：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.

参数:

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.

备注

• 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)

Initialization.

参数:

• 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)

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

参数:

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

add_exclude(exclude=None)

Add keys to exclude from inputs when creating cache entries.

参数:

exclude (Optional[Iterable[str]]) -- Keys to add to the exclusion list.

__contains__(key)

Checks if a cache entry exists for the given key.

参数:

key (Union[int, CacheEntry]) -- The key or CacheEntry to check in the cache.

返回:

True if the cache entry exists, False otherwise.

返回类型:

bool

exists(func, **kwargs)

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

参数:

• func (Union[Callable, str]) -- The function or its name to check in the cache.

• **kwargs -- Arbitrary keyword arguments representing the inputs to the function.

返回:

True if the cache entry exists, False otherwise.

返回类型:

bool

__getitem__(key)

Retrieves a cache entry for the given function and inputs.

参数:

key (Union[int, str, CacheEntry]) -- The key or CacheEntry to retrieve.

返回:

The cached entry if found. Otherwise Ellipsis.

返回类型:

CacheEntry

retrieve(func, **kwargs)

Retrieves a cached entry for the given function and inputs.

参数:

• 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.

返回:

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

返回类型:

Any

get(func, **kwargs)

Gets a cached value for the given function and inputs.

参数:

• 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.

返回:

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

返回类型:

Any

__setitem__(key, value)

Sets a cache entry for the given function and inputs.

参数:

• 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)

Sets a cached value for the given function and inputs.

参数:

• 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)

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

参数:

entry (CacheEntry) -- The cache entry to add.

annotate(func, expected=..., metadata=None, **kwargs)

Annotates a cached value for the given function and inputs.

参数:

• 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.

抛出:

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

__delitem__(key)

Deletes a cache entry for the given function and inputs.

参数:

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

unset(func, **kwargs)

Deletes a cache entry for the given function and inputs.

参数:

• 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)

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

参数:

entry (CacheEntry) -- The cache entry to remove.

__iter__()

Iterates over the cache entries.

生成器:

CacheEntry -- The CacheEntry objects in the cache.

返回类型:

Generator[CacheEntry, None, None]

pop()

Pops an arbitrary cache entry from the cache.

返回:

The popped CacheEntry if the cache is not empty, otherwise None.

返回类型:

Optional[CacheEntry]

popall()

Pops all cache entries from the cache.

返回:

A list of all popped CacheEntry objects.

返回类型:

List[CacheEntry]

clear()

Clears the cache.

close()

Closes the cache, if applicable.

flush(**kwargs)

Flushes the cache, if applicable.

memoize(func=None, *, name=None)

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

返回类型:

Callable

参数:

• 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)

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(...): ...

参数:

• func (Callable) -- The function to cache.

• name (Optional[str]) -- Optional name for the function in the cache key.

返回:

The wrapped function with batch caching.

返回类型:

Callable