Cache

persista.cache

Contain caches.

persista.cache.AsyncTTLCache

Async cache with per-entry expiry, backed by any :class:~persista.store.base.AsyncBaseStore.

Each entry is wrapped as {"value": value, "expires_at": expires_at} before being written to the store, since :class:~persista.store.base.AsyncBaseStore only accepts dict values. If the backing store is one that serializes values (e.g. a SQLite- or Redis-backed store), cached values must be JSON-serializable.

expires_at is a Unix timestamp (time.time()), not a monotonic clock reading, because entries may be read back by a different process or after a restart of this one. Expiry is checked lazily on :meth:get: an expired entry is only evicted the next time it is looked up, not proactively at its expiry time.

Parameters:

Name	Type	Description	Default
store	AsyncBaseStore | None	The backing store. Defaults to a new :class:~persista.store.async_in_memory.AsyncInMemoryStore.	None
default_ttl	float	The default time-to-live, in seconds, applied to entries whose ttl is not explicitly set. Must be positive.	300

Raises:

Type	Description
ValueError	If default_ttl is not positive.

Example

>>> import asyncio
>>> from persista.cache import AsyncTTLCache
>>> async def main():
...     cache = AsyncTTLCache(default_ttl=60)
...     await cache.set("greeting", "hello")
...     print(await cache.get("greeting"))
...
>>> asyncio.run(main())
hello

persista.cache.AsyncTTLCache.clear async

clear() -> None

Remove every entry from the cache, expired or not.

Example

>>> import asyncio
>>> from persista.cache import AsyncTTLCache
>>> async def main():
...     cache = AsyncTTLCache()
...     await cache.set("greeting", "hello")
...     await cache.clear()
...     print(await cache.get("greeting"))
...
>>> asyncio.run(main())
None

persista.cache.AsyncTTLCache.get async

get(key: str) -> Any | None

Retrieve a value by its key.

If the entry has expired, it is evicted from the backing store as a side effect of this call, before None is returned.

Parameters:

Name	Type	Description	Default
key	str	The key to look up.	required

Returns:

Type	Description
Any | None	The cached value, or None if the key is missing or
Any | None	its entry has expired. This means a cached value of
Any | None	None is indistinguishable from a cache miss.

Example

>>> import asyncio
>>> from persista.cache import AsyncTTLCache
>>> async def main():
...     cache = AsyncTTLCache()
...     await cache.set("greeting", "hello")
...     print(await cache.get("greeting"))
...     print(await cache.get("missing"))
...
>>> asyncio.run(main())
hello
None

persista.cache.AsyncTTLCache.memoize

memoize(
    ttl: float | None = None,
    strategy: str = "pickle",
    ignore_non_serializable: bool = False,
) -> Callable[
    [Callable[..., Awaitable[T]]],
    Callable[..., Awaitable[T]],
]

Decorate an async function so its return values are cached.

The cache key is derived from the decorated function's qualified name (__qualname__) and call arguments, via :func:~persista.cache.utils.make_key, so calls with equal arguments share a cached result. Call arguments must be serializable with strategy, unless ignore_non_serializable is set; the return value must additionally be JSON-serializable if the backing store serializes values (see the class docstring). Because the key is based on __qualname__ rather than object identity, two distinct functions defined with the same qualified name (e.g. two calls to the same factory returning a closure) share their cache entries.

Parameters:

Name	Type	Description	Default
ttl	float | None	The time-to-live, in seconds, applied to cached results. Defaults to self.default_ttl. Must be positive.	None
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'pickle'
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not serializable with strategy are dropped before computing the key, instead of raising an error. See :func:~persista.cache.utils.make_key.	False

Returns:

Type	Description
Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]	A decorator that wraps an async function with caching.

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> import asyncio
>>> from persista.cache import AsyncTTLCache
>>> cache = AsyncTTLCache()
>>> calls = []
>>> @cache.memoize(ttl=60)
... async def square(x):
...     calls.append(x)
...     return x * x
...
>>> async def main():
...     print(await square(4))
...     print(await square(4))  # served from the cache, not re-computed
...
>>> asyncio.run(main())
16
16
>>> calls
[4]

persista.cache.AsyncTTLCache.set async

set(key: str, value: Any, ttl: float | None = None) -> None

Add a value to the cache.

Calling this again with an existing key overwrites the previous value and resets its expiry.

Parameters:

Name	Type	Description	Default
key	str	The key to set.	required
value	Any	The value to cache. Must be JSON-serializable if the backing store serializes values (see the class docstring).	required
ttl	float | None	The time-to-live, in seconds, before the entry expires. Defaults to self.default_ttl. Must be positive.	None

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> import asyncio
>>> from persista.cache import AsyncTTLCache
>>> async def main():
...     cache = AsyncTTLCache()
...     await cache.set("greeting", "hello")
...     print(await cache.get("greeting"))
...     await cache.set("greeting", "bonjour")
...     print(await cache.get("greeting"))
...
>>> asyncio.run(main())
hello
bonjour

persista.cache.TTLCache

Cache with per-entry expiry, backed by any :class:~persista.store.base.BaseStore.

Each entry is wrapped as {"value": value, "expires_at": expires_at} before being written to the store, since :class:~persista.store.base.BaseStore only accepts dict values. If the backing store is one that serializes values (e.g. a SQLite- or Redis-backed store), cached values must be JSON-serializable.

expires_at is a Unix timestamp (time.time()), not a monotonic clock reading, because entries may be read back by a different process or after a restart of this one. Expiry is checked lazily on :meth:get: an expired entry is only evicted the next time it is looked up, not proactively at its expiry time.

Parameters:

Name	Type	Description	Default
store	BaseStore | None	The backing store. Defaults to a new :class:~persista.store.in_memory.InMemoryStore.	None
default_ttl	float	The default time-to-live, in seconds, applied to entries whose ttl is not explicitly set. Must be positive.	300

Raises:

Type	Description
ValueError	If default_ttl is not positive.

Example

>>> from persista.cache import TTLCache
>>> cache = TTLCache(default_ttl=60)
>>> cache.set("greeting", "hello")
>>> cache.get("greeting")
'hello'

persista.cache.TTLCache.clear

clear() -> None

Remove every entry from the cache, expired or not.

Example

>>> from persista.cache import TTLCache
>>> cache = TTLCache()
>>> cache.set("greeting", "hello")
>>> cache.clear()
>>> cache.get("greeting") is None
True

persista.cache.TTLCache.get

get(key: str) -> Any | None

Retrieve a value by its key.

If the entry has expired, it is evicted from the backing store as a side effect of this call, before None is returned.

Parameters:

Name	Type	Description	Default
key	str	The key to look up.	required

Returns:

Type	Description
Any | None	The cached value, or None if the key is missing or
Any | None	its entry has expired. This means a cached value of
Any | None	None is indistinguishable from a cache miss.

Example

>>> from persista.cache import TTLCache
>>> cache = TTLCache()
>>> cache.set("greeting", "hello")
>>> cache.get("greeting")
'hello'
>>> cache.get("missing") is None
True

persista.cache.TTLCache.memoize

memoize(
    ttl: float | None = None,
    strategy: str = "pickle",
    ignore_non_serializable: bool = False,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Decorate a function so its return values are cached.

Works on both sync and async functions (async def).

The cache key is derived from the decorated function's qualified name (__qualname__) and call arguments, via :func:~persista.cache.utils.make_key, so calls with equal arguments share a cached result. Call arguments must be serializable with strategy, unless ignore_non_serializable is set; the return value must additionally be JSON-serializable if the backing store serializes values (see the class docstring). Because the key is based on __qualname__ rather than object identity, two distinct functions defined with the same qualified name (e.g. two calls to the same factory returning a closure) share their cache entries.

Parameters:

Name	Type	Description	Default
ttl	float | None	The time-to-live, in seconds, applied to cached results. Defaults to self.default_ttl. Must be positive.	None
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'pickle'
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not serializable with strategy are dropped before computing the key, instead of raising an error. See :func:~persista.cache.utils.make_key.	False

Returns:

Type	Description
Callable[[Callable[..., T]], Callable[..., T]]	A decorator that wraps a function with caching.

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> from persista.cache import TTLCache
>>> cache = TTLCache()
>>> calls = []
>>> @cache.memoize(ttl=60)
... def square(x):
...     calls.append(x)
...     return x * x
...
>>> square(4)
16
>>> square(4)  # served from the cache, not re-computed
16
>>> calls
[4]

persista.cache.TTLCache.set

set(key: str, value: Any, ttl: float | None = None) -> None

Add a value to the cache.

Calling this again with an existing key overwrites the previous value and resets its expiry.

Parameters:

Name	Type	Description	Default
key	str	The key to set.	required
value	Any	The value to cache. Must be JSON-serializable if the backing store serializes values (see the class docstring).	required
ttl	float | None	The time-to-live, in seconds, before the entry expires. Defaults to self.default_ttl. Must be positive.	None

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> from persista.cache import TTLCache
>>> cache = TTLCache()
>>> cache.set("greeting", "hello")
>>> cache.get("greeting")
'hello'
>>> cache.set("greeting", "bonjour")
>>> cache.get("greeting")
'bonjour'
>>> cache.set("short-lived", "value", ttl=30)

persista.cache.async_cached

async_cached(
    ttl: int | None = None,
    strategy: str = "pickle",
    ignore_non_serializable: bool = False,
) -> Callable[
    [Callable[..., Awaitable[T]]],
    Callable[..., Awaitable[T]],
]

Cache an async function's return values in the shared default async cache.

Looks up :func:get_async_ttl_cache on every call, so replacing the shared cache via :func:set_async_ttl_cache also changes where already-decorated functions store their results.

The cache key is derived from the decorated function's qualified name (__qualname__) and call arguments, via :func:~persista.cache.utils.make_key.

Parameters:

Name	Type	Description	Default
ttl	int | None	The time-to-live, in seconds, applied to cached results. Defaults to the cache's default_ttl. Must be positive.	None
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'pickle'
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not serializable with strategy are dropped before computing the key, instead of raising an error. See :func:~persista.cache.utils.make_key.	False

Returns:

Type	Description
Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]	A decorator that wraps an async function with caching.

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> import asyncio
>>> from persista.cache import async_cached
>>> calls = []
>>> @async_cached(ttl=60)
... async def square(x):
...     calls.append(x)
...     return x * x
...
>>> async def main():
...     print(await square(4))
...     print(await square(4))  # served from the cache, not re-computed
...
>>> asyncio.run(main())
16
16
>>> calls
[4]

persista.cache.cached

cached(
    ttl: int | None = None,
    strategy: str = "pickle",
    ignore_non_serializable: bool = False,
) -> Callable[[Callable[..., T]], Callable[..., T]]

Cache a function's return values in the shared default cache.

Works on both sync and async functions (async def), by looking up :func:get_ttl_cache on every call, so replacing the shared cache via :func:set_ttl_cache also changes where already-decorated functions store their results.

The cache key is derived from the decorated function's qualified name (__qualname__) and call arguments, via :func:~persista.cache.utils.make_key.

Parameters:

Name	Type	Description	Default
ttl	int | None	The time-to-live, in seconds, applied to cached results. Defaults to the cache's default_ttl. Must be positive.	None
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'pickle'
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not serializable with strategy are dropped before computing the key, instead of raising an error. See :func:~persista.cache.utils.make_key.	False

Returns:

Type	Description
Callable[[Callable[..., T]], Callable[..., T]]	A decorator that wraps a function with caching.

Raises:

Type	Description
ValueError	If ttl is not positive.

Example

>>> from persista.cache import cached
>>> calls = []
>>> @cached(ttl=60)
... def square(x):
...     calls.append(x)
...     return x * x
...
>>> square(4)
16
>>> square(4)  # served from the cache, not re-computed
16
>>> calls
[4]

persista.cache.get_async_ttl_cache

get_async_ttl_cache() -> AsyncTTLCache

Return the shared default async cache.

Returns:

Type	Description
AsyncTTLCache	The shared default :class:~persista.cache.async_ttl.AsyncTTLCache
AsyncTTLCache	instance, used by :func:async_cached when no explicit cache
AsyncTTLCache	is given.

Example

>>> import asyncio
>>> from persista.cache.interface import get_async_ttl_cache
>>> async def main():
...     cache = get_async_ttl_cache()
...     await cache.set("greeting", "hello")
...     print(await cache.get("greeting"))
...
>>> asyncio.run(main())
hello

persista.cache.get_ttl_cache

get_ttl_cache() -> TTLCache

Return the shared default cache.

Returns:

Type	Description
TTLCache	The shared default :class:~persista.cache.ttl.TTLCache
TTLCache	instance, used by :func:cached when no explicit cache is
TTLCache	given.

Example

>>> from persista.cache.interface import get_ttl_cache
>>> cache = get_ttl_cache()
>>> cache.set("greeting", "hello")
>>> cache.get("greeting")
'hello'

persista.cache.make_json_key

make_json_key(
    func_name: str,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    ignore_non_serializable: bool = False,
) -> str

Derive a stable cache key from a function name and its call arguments.

func_name, args, and kwargs are JSON-serialized (with kwargs keys sorted, so key order doesn't affect the result) and hashed. args and the values in kwargs must be JSON-serializable, unless ignore_non_serializable is set.

Parameters:

Name	Type	Description	Default
func_name	str	The name of the function being cached, typically its __qualname__.	required
args	tuple[Any, ...]	The positional arguments the function was called with. Must be JSON-serializable, unless ignore_non_serializable is set.	required
kwargs	dict[str, Any]	The keyword arguments the function was called with. Must be JSON-serializable, unless ignore_non_serializable is set.	required
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not JSON-serializable are dropped before computing the key, instead of raising an error. This means calls that only differ in a non-serializable argument (e.g. a logger or a client instance) map to the same key.	False

Returns:

Type	Description
str	A hash of func_name, args, and kwargs, stable
str	across calls with equal arguments regardless of kwargs
str	order.

Raises:

Type	Description
TypeError	If args or kwargs contains a value that is not JSON-serializable and ignore_non_serializable is False.

Example

>>> from persista.cache.utils import make_json_key
>>> make_json_key("add", (1, 2), {}) == make_json_key("add", (1, 2), {})
True
>>> make_json_key("add", (), {"a": 1, "b": 2}) == make_json_key("add", (), {"b": 2, "a": 1})
True
>>> make_json_key("add", (1, 2), {}) == make_json_key("add", (1, 3), {})
False

persista.cache.make_key

make_key(
    func_name: str,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    strategy: str = "pickle",
    ignore_non_serializable: bool = False,
) -> str

Derive a stable cache key from a function name and its call arguments, using the given serialization strategy.

Parameters:

Name	Type	Description	Default
func_name	str	The name of the function being cached, typically its __qualname__.	required
args	tuple[Any, ...]	The positional arguments the function was called with. Must be serializable with strategy, unless ignore_non_serializable is set.	required
kwargs	dict[str, Any]	The keyword arguments the function was called with. Must be serializable with strategy, unless ignore_non_serializable is set.	required
strategy	str	The serialization strategy used to compute the key. Either "json" (see make_json_key) or "pickle" (see make_pickle_key).	'pickle'
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not serializable with strategy are dropped before computing the key, instead of raising an error. This means calls that only differ in a non-serializable argument (e.g. a logger or a client instance) map to the same key.	False

Returns:

Type	Description
str	A hash of func_name, args, and kwargs, stable
str	across calls with equal arguments regardless of kwargs
str	order.

Raises:

Type	Description
ValueError	If strategy is not "json" or "pickle".
TypeError	If strategy is "json" and args or kwargs contains a value that is not JSON-serializable and ignore_non_serializable is False.
PicklingError	If strategy is "pickle" and args or kwargs contains a value that cannot be pickled and ignore_non_serializable is False.

Example

>>> from persista.cache.utils import make_key
>>> make_key("add", (1, 2), {}, strategy="json") == make_key(
...     "add", (1, 2), {}, strategy="json"
... )
True
>>> make_key("add", (1, 2), {}) == make_key("add", (1, 3), {})
False

persista.cache.make_pickle_key

make_pickle_key(
    func_name: str,
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    ignore_non_serializable: bool = False,
) -> str

Derive a stable cache key from a function name and its call arguments.

This is similar to make_json_key but uses pickle instead of json to serialize func_name, args, and kwargs before hashing, so it supports a broader range of argument types at the cost of a key that is only stable within a single Python version (pickle's format can change across versions).

Parameters:

Name	Type	Description	Default
func_name	str	The name of the function being cached, typically its __qualname__.	required
args	tuple[Any, ...]	The positional arguments the function was called with. Must be picklable, unless ignore_non_serializable is set.	required
kwargs	dict[str, Any]	The keyword arguments the function was called with. Must be picklable, unless ignore_non_serializable is set.	required
ignore_non_serializable	bool	If True, positional arguments and keyword argument values that are not picklable are dropped before computing the key, instead of raising an error. This means calls that only differ in a non-picklable argument (e.g. a logger or a client instance) map to the same key.	False

Returns:

Type	Description
str	A hash of func_name, args, and kwargs, stable
str	across calls with equal arguments regardless of kwargs
str	order.

Raises:

Type	Description
PicklingError	If args or kwargs contains a value that cannot be pickled and ignore_non_serializable is False.

Example

>>> from persista.cache.utils import make_pickle_key
>>> make_pickle_key("add", (1, 2), {}) == make_pickle_key("add", (1, 2), {})
True
>>> make_pickle_key("add", (), {"a": 1, "b": 2}) == make_pickle_key(
...     "add", (), {"b": 2, "a": 1}
... )
True
>>> make_pickle_key("add", (1, 2), {}) == make_pickle_key("add", (1, 3), {})
False

persista.cache.set_async_ttl_cache

set_async_ttl_cache(cache: AsyncTTLCache) -> None

Replace the shared default async cache.

Parameters:

Name	Type	Description	Default
cache	AsyncTTLCache	The :class:~persista.cache.async_ttl.AsyncTTLCache instance to install as the new shared default, in place of the one returned by :func:get_async_ttl_cache.	required

Example

>>> from persista.cache import AsyncTTLCache
>>> from persista.cache import get_async_ttl_cache, set_async_ttl_cache
>>> set_async_ttl_cache(AsyncTTLCache(default_ttl=60))
>>> get_async_ttl_cache().default_ttl
60

persista.cache.set_ttl_cache

set_ttl_cache(cache: TTLCache) -> None

Replace the shared default cache.

Parameters:

Name	Type	Description	Default
cache	TTLCache	The :class:~persista.cache.ttl.TTLCache instance to install as the new shared default, in place of the one returned by :func:get_ttl_cache.	required

Example

>>> from persista.cache import TTLCache
>>> from persista.cache import get_ttl_cache, set_ttl_cache
>>> set_ttl_cache(TTLCache(default_ttl=60))
>>> get_ttl_cache().default_ttl
60