anystore.decorators

Decorate functions to store results in a configurable cache and retrieve cached results on next call.

Example

from anystore import anycache

@anycache(uri="./local-cache")
def calculate(data):
    # a very time consuming task
    return result

# 1. time: slow
res = calculate(100)

# 2. time: fast, as now cached
res = calculate(100)

The cache key is computed based on the input arguments, but can be configured.

See below for reference details.

anycache(func=None, store=None, model=None, key_func=None, serialization_mode='auto', serialization_func=None, deserialization_func=None, ttl=None, **store_kwargs)

Cache a function call in a configurable cache backend. By default, the default store is used (configured via environment)

Example

@anycache(
    store=get_store("redis://localhost"),
    key_func=lambda *args, **kwargs: args[0].upper()
)
def compute(*args, **kwargs):
    return "result"

Note

If the key_func returns None as the computed cache key, the result will not be cached (this can be used to dynamically disable caching based on function input)

See anystore.serialize for serialization reference.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any] | None	The function to wrap	None
serialization_mode	Mode | None	"auto", "pickle", "json", "raw"	'auto'
serialization_func	Callable | None	Function to use to serialize	None
deserialization_func	Callable | None	Function to use to deserialize, takes bytes as input	None
model	BaseModel | None	Pydantic model to use for serialization from a json bytes string	None
key_func	Callable[..., str | None] | None	Function to compute the cache key	None
ttl	int | None	Key ttl for supported backends	None
**store_kwargs	Any	Any other store options or backend specific configuration to pass through	{}

Returns:

Name	Type	Description
Callable	Callable[..., Any]	The decorated function

Source code in anystore/decorators.py

def anycache(
    func: Callable[..., Any] | None = None,
    store: BaseStore | None = None,
    model: BaseModel | None = None,
    key_func: Callable[..., str | None] | None = None,
    serialization_mode: Mode | None = "auto",
    serialization_func: Callable | None = None,
    deserialization_func: Callable | None = None,
    ttl: int | None = None,
    **store_kwargs: Any
) -> Callable[..., Any]:
    """
    Cache a function call in a configurable cache backend. By default, the
    default store is used (configured via environment)

    Example:
        ```python
        @anycache(
            store=get_store("redis://localhost"),
            key_func=lambda *args, **kwargs: args[0].upper()
        )
        def compute(*args, **kwargs):
            return "result"
        ```

    Note:
        If the `key_func` returns `None` as the computed cache key, the result
        will not be cached (this can be used to dynamically disable caching
        based on function input)

    See [`anystore.serialize`][anystore.serialize] for serialization reference.

    Args:
        func: The function to wrap
        serialization_mode: "auto", "pickle", "json", "raw"
        serialization_func: Function to use to serialize
        deserialization_func: Function to use to deserialize, takes bytes as input
        model: Pydantic model to use for serialization from a json bytes string
        key_func: Function to compute the cache key
        ttl: Key ttl for supported backends
        **store_kwargs: Any other store options or backend specific
            configuration to pass through

    Returns:
        Callable: The decorated function
    """
    key_func, store = _setup_decorator(
        store=store,
        key_func=key_func,
        serialization_mode=serialization_mode,
        serialization_func=serialization_func,
        deserialization_func=deserialization_func,
        model=model,
        ttl=ttl,
        **store_kwargs
    )

    def _decorator(func):
        @functools.wraps(func)
        def _inner(*args, **kwargs):
            key = key_func(*args, **kwargs)
            try:
                if key is not None:
                    return store.get(key)
                raise DoesNotExist
            except DoesNotExist:
                res = func(*args, **kwargs)
                return _handle_result(key, res, store)

        return _inner

    if func is None:
        return _decorator
    return _decorator(func)

async_anycache(func=None, **store_kwargs)

Async implementation of the @anycache decorator

Source code in anystore/decorators.py

def async_anycache(func=None, **store_kwargs):
    """
    Async implementation of the [@anycache][anystore.decorators.anycache]
    decorator
    """
    key_func, store = _setup_decorator(**store_kwargs)

    def _decorator(func):
        @functools.wraps(func)
        async def _inner(*args, **kwargs):
            key = key_func(*args, **kwargs)
            try:
                if key is not None:
                    return store.get(key)
                raise DoesNotExist
            except DoesNotExist:
                res = await func(*args, **kwargs)
                return _handle_result(key, res, store)

        return _inner

    if func is None:
        return _decorator
    return _decorator(func)