redis_func_cache.cache module

redis_func_cache.cache module#

class redis_func_cache.cache.RedisFuncCache(name, policy, *, client=None, factory=None, maxsize=1024, ttl=3600, update_ttl=True, prefix='func-cache:', serializer='json')[source]#

Bases: Generic[RedisClientTV]

A function cache class backed by Redis.

This class provides a decorator-based caching mechanism for functions, storing their results in Redis. It supports both synchronous and asynchronous Redis clients, customizable cache policies, and flexible serialization options.

Example

import redis
from redis_func_cache import LruTPolicy, RedisFuncCache

pool = redis.ConnectionPool(...)
factory = redis.from_pool(pool)

# supply a client instance by a factory
cache = RedisFuncCache(__name__, LruTPolicy(), factory=factory)

@cache
def function_to_cache(...):
    ...

  • The serializer parameter can be a string or a pair of callables.

  • If a callable factory is provided via the factory parameter, it will be invoked every time the redis client is accessed.

  • The cache supports both synchronous and asynchronous Redis clients, but the decorated function must match the redis client’s async/sync nature.

Initializes the Cache instance with the given parameters.

Parameters:

  • name (str) –

    The name of the cache manager.

    It is assigned to property name.

  • policy (AbstractPolicy) –

    A pre-instantiated AbstractPolicy instance to use for

    eviction and key/hash calculation.

    The provided policy instance will be bound to this cache by setting its internal cache reference to a weakref proxy of this cache. If you need a fresh policy instance per cache, create a new policy object and pass it here. Reusing the same policy instance across multiple caches is discouraged as policies commonly hold cache-specific state.

    Changed in version 0.7: The policy argument now accepts a pre-instantiated policy instance, NOT a class.

  • client (TypeVar(RedisClientTV, bound= Redis | Redis | RedisCluster | RedisCluster) | None) –

    Optional Redis client instance to use.

    This argument may be an already-created Redis client instance (for simple scripts/tests), or None when a factory is supplied.

    Examples of a client instance:

    Changed in version 0.7: Prefer providing a factory for concurrent/production use; use client only for simple cases or compatibility.

  • factory (Callable[[], TypeVar(RedisClientTV, bound= Redis | Redis | RedisCluster | RedisCluster)] | None) –

    Optional callable that returns a Redis client instance.

    If provided, the callable will be invoked every time get_client() is called or the cache instance requires a redis client internally. When both factory and client are provided, factory takes precedence and will be used to obtain clients.

    Added in version 0.7.

  • maxsize (int) –

    The maximum size of the cache.

    Assigned to property maxsize.

  • ttl (int) –

    The time-to-live (in seconds) for the whole cache data structures on Redis backend.

    Assigned to property ttl.

  • update_ttl (bool) –

    Whether to update the TTL of the whole cache data structures on Redis backend when they are accessed.

    • When True (default), accessing a cached item will reset its TTL.

    • When False, accessing a cached item will not update its TTL.

    Assigned to property update_ttl.

    Added in version 0.5.

  • prefix (str) –

    The prefix for cache keys.

    If not provided, the default is DEFAULT_PREFIX.

    Assigned to property prefix.

  • serializer (Literal['json', 'pickle', 'dill', 'bson', 'msgpack', 'yaml', 'cbor', 'cloudpickle'] | tuple[Callable[[Any], bytes | bytearray | memoryview], Callable[[bytes | bytearray | memoryview], Any]]) –

    Optional serialize/deserialize name or function pair for return value of what decorated.

    The decorated function‘s return value is serialized to string or bytes then stored in Redis when cached, deserialized to a Python object when retrieved.

    If not provided, the cache will use json.dumps() and json.loads().

    • It could be a string, and must be one of the following:

    • Or it could be a PAIR of callbacks, the first one is used to serialize return value, the second one is used to deserialize return value.

      Here is an example of first(serialize) callback:

      def my_serializer(value):
    return msgpack.packb(value, use_bin_type=True)

      And and example of second(deserialize) callback:

      def my_deserializer(data):
    return msgpack.unpackb(data, raw=False)

      We can then pass the two callbacks to serializer parameter:

      my_cache = RedisFuncCache(
    __name__,
    MyPolicy(),
    client=redis_client,
    # here pass two callbacks to serializer
    serializer=(my_serializer, my_deserializer),
    # ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
)

    This argument is assigned to property serializer.

__call__#

Equivalent to the decorate() method.

__serializers__#

A dictionary of serializers.

Type:

dict[str, SerializerPairT]

class Mode(read=True, write=True, exec=True)[source]#

Bases: object

A dataclasses.dataclass() for cache operation mode flags.

Defines how the cache behaves when executing decorated functions.

Added in version 0.5.

Parameters:
exec: bool = True#

Allow function execution

read: bool = True#

Allow reading from cache backend

write: bool = True#

Allow writing to cache backend

class Stats(count=0, read=0, write=0, exec=0, miss=0, hit=0)[source]#

Bases: object

A dataclasses.dataclass() for cache operation statistics.

Added in version 0.5.

Parameters:
count: int = 0#

Number of cache operations

exec: int = 0#

Number of function execution operations

hit: int = 0#

Number of cache hits

miss: int = 0#

Number of cache misses

read: int = 0#

Number of read operations

write: int = 0#

Number of write operations

async aexec(user_function, user_args, user_kwds, serialize_func=None, deserialize_func=None, bound=None, field_ttl=0, **options)[source]#

Asynchronous version of exec()

Return type:

Any

Parameters:
async classmethod aget(script, keys, hash_, update_ttl, ttl, options=None, ext_args=None)[source]#

Async version of get()

Return type:

bytes | bytearray | memoryview | None

Parameters:
async classmethod aput(script, keys, hash_, value, maxsize, update_ttl, ttl, field_ttl=0, options=None, ext_args=None)[source]#

Async version of put()

Parameters:
property client: RedisClientTV#

Equivalent to call get_client().

Deprecated since version 0.5: use get_client() instead.

decorate(user_function=None, /, *, serializer=None, ttl=None, excludes=None, excludes_positional=None, **options)[source]#

Decorate the given function with caching.

Parameters:

  • user_function (TypeVar(CallableTV, bound= Callable) | None) – The function to be decorated.

  • serializer (Literal['json', 'pickle', 'dill', 'bson', 'msgpack', 'yaml', 'cbor', 'cloudpickle'] | tuple[Callable[[Any], bytes | bytearray | memoryview], Callable[[bytes | bytearray | memoryview], Any]] | None) –

    serialize/deserialize for the return value.

    Serializer/deserializer name or function pair for the decorated function’s return value. The decorated function’s return value is serialized to string or bytes and then stored in Redis when cached, and deserialized back to a Python object when retrieved.

    It accepts either:

    • A string key mapping to predefined serializers (like "yaml", "json")

    • A tuple of (serialize_func, deserialize_func) functions

    Note

    If assigned, it overwrite the serializer property of the cache instance on, and only affect the currently decorated function.

  • ttl (int | None) –

    (Experimental) The time-to-live (in seconds) for a single invocation’s result cache.

    Note

    This parameter specifies the expiration time for a single invocation result inside the cached structures, not for the entire cache.

    Caution

    This experimental expiration mechanism relies on Redis Hashes Field expiration. Expiration only applies to the HASH field (the cached return value), and does not reduce the total number of items in the cache when a field expires.

    Typically, the cached return value in the HASH portion is automatically released after expiration. However, the corresponding hash key in the ZSET portion is not removed automatically. Instead, it is only “lazily” cleaned up when accessed, or removed by the eviction policy when a new value is added. During this period, the ZSET portion continues to occupy memory, and the reported number of cache items does not decrease.

    Warning

    This feature is experimental and requires Redis 7.4 or above.

    Added in version 0.5.

  • excludes (Sequence[str] | None) –

    Optional sequence of parameter names specifying keyword arguments to exclude from cache key generation.

    Example

    @cache(excludes=["session", "token"])
def update_user(user_id: int, session: Session, token: str) -> None: ...

    Added in version 0.5.

  • excludes_positional (Sequence[int] | None) –

    Optional sequence of indices specifying positional arguments to exclude from cache key generation.

    Example

    @cache(excludes_positional=[1, 2])
def update_user(user_id: int, session: Session, token: str) -> None: ...

    options: Additional options passed to exec(), they will encoded to json, then pass to redis lua script.

    Added in version 0.5.

Return type:

TypeVar(CallableTV, bound= Callable)

This method is equivalent to __call__.

Example

Once we create a cache instance:

from redis_func_cache import RedisFuncCache

cache = RedisFuncCache("my_cache", MyPolicy(), client=redis_client)

We can use it as a decorator, either the instance itself or the decorate() method, with or without parentheses:

@cache
def my_func(a, b):
    return a + b

or:

@cache()
def my_func(a, b):
    return a + b

or:

@cache.decorate
def my_func(a, b):
    return a + b

or:

@cache.decorate()
def my_func(a, b):
    return a + b

or:

@cache(serializer=(lambda x: yaml.safe_dump(x), lambda y: yaml.safe_load(y)))
def my_func(a, b):
    return a + b

or:

@cache.decorate(serializer="yaml")
def my_func(a, b):
    return a + b
deserialize(data, f=None)[source]#

Deserialize the return value of the decorated function.

Parameters:
Return type:

Any

Returns:

The deserialized value.

disable_rw()[source]#

A context manager who disables the cache read and write temporarily.

This wall disable the cache read and write, as if there is no cache.

Example

@cache
def func(): ...


with cache.disable_rw():
    result = func()  # will be executed without cache ability

Added in version 0.5.

Return type:

Generator[Mode]

exec(user_function, user_args, user_kwds, serialize_func=None, deserialize_func=None, bound=None, field_ttl=0, **options)[source]#

Execute the given user function with the provided arguments.

Parameters:

  • user_function (Callable) – The user function to execute.

  • user_args (tuple[Any, ...]) – Positional arguments to pass to the user function.

  • user_kwds (dict[str, Any]) – Keyword arguments to pass to the user function.

  • serialize_func (Callable[[Any], bytes | bytearray | memoryview] | None) – Custom serializer passed from decorate().

  • deserialize_func (Callable[[bytes | bytearray | memoryview], Any] | None) – Custom deserializer passed from decorate().

  • bound (BoundArguments | None) –

    Filtered bound arguments which will be used by the policy of the cache.

    • If it is provided, the policy will only use the filtered arguments to calculate the cache key and hash value.

    • If it is not provided, the policy will use all arguments to calculate the cache key and hash value.

  • field_ttl (int) – Time-to-live (in seconds) for the cached field.

  • options – Additional options from decorate()’s **kwargs.

Return type:

Any

Returns:

The cached return value if it exists in the cache; otherwise, the direct return value of the user function.

Notes

This method first calls get() to attempt retrieving a cached result before executing the user_function. If no cached result is found, it executes the user_function, then calls put() to store the result in the cache.

classmethod get(script, keys, hash_value, update_ttl, ttl, options=None, ext_args=None)[source]#

Execute the given Redis Lua script with the provided arguments.

Parameters:

  • script (Script) – Redis Lua script to be evaluated, which should attempt to retrieve the return value from the cache using the given keys and hash.

  • key_pair – The key name pair of the Redis set and hash-map data structure used by the cache.

  • hash_value (bytes | str | memoryview) – The member of the Redis key and also the field name of the Redis hash map.

  • ttl (int) – Time-to-live of the cache in seconds.

  • options (Mapping[str, Any] | None) – Reserved for future use.

  • ext_args (Iterable[bytes | bytearray | memoryview | str | int | float] | None) – Extra arguments passed to the Lua script.

  • keys (tuple[bytes | str | memoryview, bytes | str | memoryview])

  • update_ttl (bool)

Return type:

bytes | bytearray | memoryview | None

Returns:

The hit return value, or None if the value is missing.

get_client()[source]#

Get the redis client instance used in the cache.

Return type:

TypeVar(RedisClientTV, bound= Redis | Redis | RedisCluster | RedisCluster)

Returns:

The redis client instance used in the cache.

  • If a redis client instance was passed to the cache constructor, it will be returned.

  • If a factory callable was provided to the constructor, the factory will be invoked

    and its result returned.

Caution:

When using the factory pattern (providing a factory callable), the factory function will be invoked every time when this method is called.

Thus the cache class will call the factory every time it needs a client internally.

Added in version 0.5.

get_mode()[source]#

Return a copy of the cache mode.

The cache mode inside the class instance is contextual variable, which is thread local and thread safe.

Added in version 0.5.

Return type:

Mode

classmethod make_bound(user_func, user_args, user_kwds, excludes=None, excludes_positional=None)[source]#
Return type:

BoundArguments | None

Parameters:
property maxsize: int#

The cache’s maximum capacity size.

mode_context(mode)[source]#

A context manager to control cache behavior.

This context manager allows you to temporarily change cache mode flags. It’s useful for temporarily disabling specific cache capabilities (e.g., disabling cache writes while keeping reads enabled).

The context manager is thread local and thread safe.

Parameters:

mode (Mode) – The cache mode to use within the context. Can be a combination of Mode bitwise flags.

Return type:

Generator[Mode]

Example

@cache
def func(): ...


mode = cache.get_mode()
mode.write = False

with cache.mode_context(mode):
    result = func()  # will be executed without cache write

Added in version 0.5.

property name: str#

The name of the cache instance.

property policy: AbstractPolicy#

Instance of the caching policy.

Note

The property returns the policy instance bound to this cache.

property prefix: str#

The prefix for cache keys.

prepare(user_function, user_args, user_kwds, bound=None)[source]#
Return type:

tuple[tuple[bytes | str | memoryview, bytes | str | memoryview], bytes | str | memoryview, Iterable[bytes | bytearray | memoryview | str | int | float]]

Parameters:
classmethod put(script, keys, hash_value, value, maxsize, update_ttl, ttl, field_ttl=0, options=None, ext_args=None)[source]#

Execute the given Redis Lua script with the provided arguments.

Parameters:

If the cache reaches its maxsize, it will remove one item according to its policy before inserting the new item.

read_only()[source]#

A context manager that only reads from cache but does not write to cache.

It doesn’t execute the function if the cache was hit.

A CacheMissError will be raised if the cache was missed in readonly mode.

Example

@cache
def func(): ...


with cache.read_only():
    # `func()` will NOT be executed, result is read from cache, and dose not write to cache
    result = func()

Added in version 0.5.

Return type:

Generator[Mode]

reset_mode(token)[source]#

Added in version 0.5.

Parameters:

token (Token)

serialize(value, f=None)[source]#

Serialize the return value of the decorated function.

The decorated function’s return value is serialized to string or bytes and then stored in Redis when cached, and deserialized back to a Python object when retrieved.

Parameters:
Return type:

bytes | bytearray | memoryview

Returns:

The serialized value.

property serializer: tuple[Callable[[Any], bytes | bytearray | memoryview], Callable[[bytes | bytearray | memoryview], Any]]#

The serializer and deserializer pair used by the cache.

This property returns a tuple of two callable objects:

  1. The first callable is the serializer, which converts data into a storable format.

  2. The second callable is the deserializer, which reconstructs the original data from the stored format.

These are used to serialize and deserialize the return values of decorated functions.

set_mode(mode)[source]#

Added in version 0.5.

Return type:

Token[Mode]

Parameters:

mode (Mode)

stats_context(stats=None)[source]#

A context manager that yields a Stats object.

Parameters:

stats (Stats | None) – The initial object to make the statistics. If None, a new stats object will be created.

Return type:

Generator[Stats]

Added in version 0.5.

Return type:

Generator[Stats]

Parameters:

stats (Stats | None)

property ttl: int#

The time-to-live (in seconds) of the whole cache structure.

property update_ttl: bool#

Whether to update the TTL of whole cache structure each time accessed.

write_only()[source]#

A context manager that bypasses cache retrieval but still writes the result to cache.

Example

@cache
def func(): ...


with cache.disable_read():
    # `func()` will be executed and result stored in cache, but not read from cache
    result = func()

Added in version 0.5.

Return type:

Generator[Mode]