API Reference

cachetoolz.decorator.Cache

Caches a function call and stores it in the namespace.

Bare decorator, @cache, is supported as well as a call with keyword arguments @cache(ttl=7200).

Parameters:

Name	Type	Description	Default
backend	Union[AsyncBackendABC, BackendABC]	Cache backend	required

Examples:

With redis async backend

>>> from cachetoolz import AsyncRedisBackend, Cache
>>> cache = Cache(AsyncRedisBackend())

With redis sync backend

>>> from cachetoolz import RedisBackend, Cache
>>> cache = Cache(RedisBackend())

@cache

Decorator for caching a function call.

Parameters:

Name	Type	Description	Default
ttl	int | float | timedelta	cache ttl (time to live)	math.inf
namespace	str	namespace to cache	'default'
typed	bool	If typed is set to true, function arguments of different types will be cached separately	False
keygen	Optional[KeyGenerator]	function to generate a cache identifier key	None

Examples:

A simple cache

>>> @cache
... def func(*args, **kwargs):
...     ...
...

Specific a namespace

>>> @cache(namespace='bar')
... def func(*args, **kwargs):
...     ...
...

Set an expiration time in seconds

>>> @cache(ttl=60)
... def func(*args, **kwargs):
...     ...
...

Use timedelta to set the expiration

>>> from datetime import timedelta
>>> @cache(ttl=timedelta(days=1))
... def func(*args, **kwargs):
...     ...
...

Differentiate caching based on argument types

>>> @cache(typed=True)
... def func(*args, **kwargs):
...     ...
...

Using a custom keygen

>>> def custom_keygen(
...     typed: bool, func: Func, *args: P.args, **kwargs: P.kwargs
... ) -> str:
...     '''Build a key to a function.
...
...     Parameters
...     ----------
...     typed
...         If typed is set to true, function arguments of different types
...         will be cached separately
...     func
...         Function
...     args
...         Function positional arguments
...     kwargs
...         Named function arguments
...
...     Returns
...     -------
...         Cache identifier key
...
...     '''
...
>>> @cache(keygen=custom_keygen)
... def func(*args, **kwargs):
...     ...
...

clear

Clears all caches for all namespaces.

This decorator will clear all caches contained in the specified namespaces once the decorated function is executed

Parameters:

Name	Type	Description	Default
namespaces	Sequence[str]	namespace to be cleaned.	('default',)

Examples:

A simple clear cache

>>> @cache.clear
... def func(*args, **kwargs):
...     ...

Defining the namespaces to be cleaned up

>>> @cache.clear(namespaces=['foo'])
... def func(*args, **kwargs):
...     ...

cachetoolz.backend

Module interface.

AsyncInMemory

Bases: AsyncBackendABC

Async in memory backend.

This backend is used to store caches in memory asynchronous.

clear(namespace) async

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key) async

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at) async

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

AsyncMongoBackend

Bases: AsyncBackendABC

Async MongoDB cache.

This backend is used to store caches mongo asynchronous.

Parameters:

Name	Type	Description	Default
host	str	MongoDB URI.	'localhost'
database	str	Cache database name.	'.cachetoolz'
kwargs	dict[str, Any]	Takes the same constructor arguments as pymongo.mongo_client.MongoClient.	{}

clear(namespace) async

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key) async

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at) async

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

AsyncRedisBackend

Bases: AsyncBackendABC

Async Redis backend.

This backend is used to store caches redis asynchronous.

Parameters:

Name	Type	Description	Default
url	str	Redis url.	required
kwargs	dict[str, Any]	Takes the same constructor arguments as redis.asyncio. client.Redis.from_url. The decode_responses parameter will always be True as the result needs to be returned as a string.	{}

clear(namespace) async

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key) async

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at) async

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

InMemory

Bases: BackendABC

In memory backend.

This backend is used to store caches in memory synchronous.

clear(namespace)

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key)

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at)

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

MongoBackend

Bases: BackendABC

MongoDB cache.

This backend is used to store caches mongo synchronous.

Parameters:

Name	Type	Description	Default
host	str	MongoDB URI.	'localhost'
database	str	Cache database name.	'.cachetoolz'
kwargs	dict[str, Any]	Takes the same constructor arguments as pymongo.mongo_client.MongoClient.	{}

clear(namespace)

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key)

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at)

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

RedisBackend

Bases: BackendABC

Redis cache.

Parameters:

Name	Type	Description	Default
url	str	Redis url.	required
kwargs	dict[str, Any]	Takes the same constructor arguments as redis.client.Redis.from_url. The decode_responses parameter will always be True as the result needs to be returned as a string.	{}

clear(namespace)

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key)

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at)

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

cachetoolz.coder

Module interface.

Coder

Bases: CoderABC

Coder class.

decode(value)

Decode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to decode.	required

Returns:

Name	Type	Description
decoded	Any	Value decoded.

encode(value)

Encode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to encode.	required

Returns:

Name	Type	Description
encoded	Any	Value encoded.

register(serializer) staticmethod

Register a JSON serializer class.

You can register a class for decoding, it needs to have the encode and decode methods where the encode method must have a parameter called value and must have the type annotated. These methods can be instance, @staticmethod, or @classmethod. The decode function will receive the exact value that is returned by the encode function.

Parameters:

Name	Type	Description	Default
class_	Union[Type[SerializerABC], SerializerABC]	Serializer class.	required

Examples:

Class methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     @classmethod
...     def encode(cls, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     @classmethod
...     def decode(cls, value):
...         return deque(val['iterable'], val['maxlen'])
...

Static methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     @staticmethod
...     def encode(value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     @staticmethod
...     def decode(value):
...         return deque(val['iterable'], val['maxlen'])
...

Instace methods

>>> from collections import deque
>>> @coder.register
>>> class DequeSerializer:
...     def encode(self, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     def decode(self, value):
...         return deque(val['iterable'], val['maxlen'])
...

When registering a class, it will be instantiated. Therefore, if the class requires any initialization parameters, you can register an instance of it along with the necessary parameters.

>>> from collections import deque
>>> class DequeCoder:
...     def __init__(self, foo):
...         self.foo = foo
...
...      def encode(self, value: deque):
...         return {'iterable': list(value), 'maxlen': value.maxlen}
...
...     def decode(self, value):
...         return deque(val['iterable'], val['maxlen'])
...
>>> coder.register(DequeCoder(foo='bar'))

cachetoolz.coder.encoder

Encoder module.

Encoder

Bases: JSONEncoder

JSON encoder class.

register(name)

Register a encoder.

Parameters:

Name	Type	Description	Default
name	str	Encoder name.	required

Examples:

>>> from collections import deque
>>> @register('deque')
... def _(value: deque):
...     return {'iterable': list(value), 'maxlen': value.maxlen}
...

cachetoolz.coder.decoder

Decoder module.

Decoder

Bases: JSONDecoder

JSON decoder class.

register(name)

Register a decoder.

Parameters:

Name	Type	Description	Default
name	str	Decoder name.	required

Examples:

>>> from collections import deque
>>> from typing import Any
>>> @register('deque')
... def _(val: dict[str, Any]):
...     return deque(val['iterable'], val['maxlen'])
...

cachetoolz.abc

Abstract module interface.

AsyncBackendABC

Bases: BaseBackend, ABC

Abstract async backend.

Attributes:

Name	Type	Description
logger	Logger	Package logger.

clear(namespace) abstractmethod async

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key) abstractmethod async

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at) abstractmethod async

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

BackendABC

Bases: BaseBackend, ABC

Abstract backend.

Attributes:

Name	Type	Description
logger	Logger	Package logger.

clear(namespace) abstractmethod

Clear a namespace.

Parameters:

Name	Type	Description	Default
namespaces	str	namespace to cache.	required

get(key) abstractmethod

Get a value if not expired.

Parameters:

Name	Type	Description	Default
key	str	Cache identifier key.	required

Returns:

Name	Type	Description
with_cache	Any	Value cached.
without_cache	None	If not exists or expired.

set(key, value, expires_at) abstractmethod

Set a value with expires time.

Parameters:

Name	Type	Description	Default
key	str	cache identifier key.	required
value	str	value to cache encoded.	required
expires_at	timedelta	expiry time.	required

CoderABC

Bases: ABC

Abstract coder.

decode(value) abstractmethod

Decode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to decode.	required

Returns:

Name	Type	Description
decoded	Any	Value decoded.

encode(value) abstractmethod

Encode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to encode.	required

Returns:

Name	Type	Description
encode	Any	Value encoded.

SerializerABC

Bases: ABC

Abstract coder.

decode(value) abstractmethod

Decode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to decode	required

Returns:

Name	Type	Description
decoded	Any	Value decoded

encode(value) abstractmethod

Encode value.

Parameters:

Name	Type	Description	Default
value	Any	Value to encode	required

Returns:

Name	Type	Description
encoded	Any	Value encoded