Cache

persista.cache

Contain caches.

persista.cache.MISSING module-attribute

MISSING: Any = object()

Sentinel usable as the default argument to :meth:Cache.get / :meth:Cache.aget to distinguish a cache miss from a cached None.

persista.cache.Cache

Bases: MultilineDisplayMixin

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

Most methods have both a sync form (get, try_get, set, contains, get_many, try_get_many, set_many, contains_many, delete, delete_many, get_or_compute, memoize, clear) and an async counterpart prefixed with a (aget, atry_get, aset, acontains, aget_many, atry_get_many, aset_many, acontains_many, adelete, adelete_many, aget_or_compute, amemoize, aclear), so the same cache instance can be used from sync and async code, provided the backing store supports the interface being used.

Each entry is wrapped as {"value": value, "expires_at": expires_at} before being written to the store, since :class:~persista.store.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.

Like :class:~persista.store.BaseStore, constructing a :class:Cache does not connect to the underlying backing store: this is deferred to :meth:open/:meth:aopen, so every other method raises until the cache has been opened, either explicitly or via use as a sync context manager (with Cache(...) as cache: ..., calling :meth:open on entry and :meth:close on exit) or an async context manager (async with Cache(...) as cache: ..., calling :meth:aopen on entry and :meth:aclose on exit).

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 | None	The default time-to-live, in seconds, applied to entries whose ttl is not explicitly set on :meth:set / :meth:get_or_compute / :meth:memoize. None (the default) means entries never expire unless an explicit ttl is given. Must be non-negative.	None

Raises:

Type	Description
ValueError	If default_ttl is negative.

Example

>>> from persista.cache import Cache
>>> with Cache(default_ttl=60) as cache:
...     cache.set("greeting", "hello")
...     cache.get("greeting")
...
'hello'

persista.cache.Cache.closed property

closed: bool

Indicate whether the cache's backing store is closed.

Returns:

Type	Description
bool	True if the backing store has been closed (or never
bool	opened), False if it is open and ready to use.

persista.cache.Cache.default_ttl property

default_ttl: float | None

The default time-to-live, in seconds, applied to entries whose ttl is not explicitly set on :meth:set / :meth:get_or_compute / :meth:memoize.

persista.cache.Cache.aclear async

aclear() -> None

Remove every entry from the cache, expired or not.

This is the async counterpart of :meth:clear, for use with an async backing store.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("greeting", "hello")
...         await cache.aclear()
...         print(await cache.aget("greeting"))
...
>>> asyncio.run(main())
None

persista.cache.Cache.aclose async

aclose() -> None

Async equivalent of :meth:close.

persista.cache.Cache.acontains async

acontains(key: str) -> bool

Indicate whether a key is present and unexpired.

This is the async counterpart of :meth:contains, for use with an async backing store.

Parameters:

Name	Type	Description	Default
key	str	The key to check.	required

Returns:

Type	Description
bool	True if key has an entry in the cache that has not
bool	expired, otherwise False. If the entry has expired,
bool	it is evicted from the backing store as a side effect of
bool	this call, as in :meth:aget.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("greeting", "hello")
...         print(await cache.acontains("greeting"))
...         print(await cache.acontains("missing"))
...
>>> asyncio.run(main())
True
False

persista.cache.Cache.acontains_many async

acontains_many(keys: list[str]) -> list[bool]

Check presence of multiple keys in a single batched store lookup.

This is the async counterpart of :meth:contains_many, for use with an async backing store.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to check.	required

Returns:

Type	Description
list[bool]	A list of booleans, in the same order as keys, where
list[bool]	each entry is True if the corresponding key is a hit
list[bool]	-- present and unexpired -- and False otherwise.
list[bool]	Expired entries are evicted from the backing store as a
list[bool]	side effect of this call, as in :meth:aget_many.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("a", "hello")
...         print(await cache.acontains_many(["a", "b"]))
...
>>> asyncio.run(main())
[True, False]

persista.cache.Cache.adelete async

adelete(key: str) -> None

Remove a single entry from the cache, if present.

This is the async counterpart of :meth:delete, for use with an async backing store.

Parameters:

Name	Type	Description	Default
key	str	The key to remove.	required

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("greeting", "hello")
...         await cache.adelete("greeting")
...         print(await cache.aget("greeting"))
...
>>> asyncio.run(main())
None

persista.cache.Cache.adelete_many async

adelete_many(keys: list[str]) -> None

Remove multiple entries from the cache in a single batched store write, if present.

This is the async counterpart of :meth:delete_many, for use with an async backing store.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to remove.	required

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset_many({"a": "hello", "b": "world"})
...         await cache.adelete_many(["a", "b"])
...         print(await cache.atry_get_many(["a", "b"]))
...
>>> asyncio.run(main())
{}

persista.cache.Cache.aget async

aget(key: str, default: Any = None) -> Any

Retrieve a value by its key.

This is the async counterpart of :meth:get, for use with an async backing store.

Parameters:

Name	Type	Description	Default
key	str	The key to look up.	required
default	Any	The value to return if the key is missing or its entry has expired. Pass the :data:~persista.cache.cache.MISSING sentinel here to distinguish a cache miss from a cached None.	None

Returns:

Type	Description
Any	The cached value, or default on a miss.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("greeting", "hello")
...         print(await cache.aget("greeting"))
...
>>> asyncio.run(main())
hello

persista.cache.Cache.aget_many async

aget_many(
    keys: list[str], default: Any = None
) -> dict[str, Any]

Retrieve multiple values in a single batched store lookup.

This is the async counterpart of :meth:get_many, for use with an async backing store.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to look up.	required
default	Any	The value to map a key to if it is missing or its entry has expired. Pass the :data:~persista.cache.cache.MISSING sentinel here to distinguish a cache miss from a cached None.	None

Returns:

Name	Type	Description
	dict[str, Any]	A dict mapping every key in keys to its cached value.
See	dict[str, Any]	meth:get_many for the exact hit/miss semantics.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("a", "hello")
...         await cache.aset("b", "world")
...         print(sorted((await cache.aget_many(["a", "b", "missing"])).items()))
...
>>> asyncio.run(main())
[('a', 'hello'), ('b', 'world'), ('missing', None)]

persista.cache.Cache.aget_or_compute async

aget_or_compute(
    key: str,
    fn: Callable[..., T] | Callable[..., Awaitable[T]],
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    ttl: float | None = _UNSET,
) -> T

Return the cached value for key, computing and storing it on a cache miss.

This is the async counterpart of :meth:get_or_compute, for use with an async backing store. fn may be a regular sync function or an async def function; either way, the backing store is always accessed through await.

Parameters:

Name	Type	Description	Default
key	str	The key to look up and, on a miss, store the result under.	required
fn	Callable[..., T] | Callable[..., Awaitable[T]]	The sync or async function to call to compute the value when key is not in the cache.	required
args	tuple[Any, ...]	Positional arguments passed to fn on a miss.	required
kwargs	dict[str, Any]	Keyword arguments passed to fn on a miss.	required
ttl	float | None	The time-to-live, in seconds, applied when storing a freshly computed value. See :meth:aset.	_UNSET

Returns:

Type	Description
T	The cached value on a hit, otherwise the value returned by
T	fn(*args, **kwargs) (awaited if fn is async).

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> calls = []
>>> async def compute(x):
...     calls.append(x)
...     return x * 2
...
>>> async def main():
...     async with Cache() as cache:
...         print(await cache.aget_or_compute("key", compute, (4,), {}))
...         print(await cache.aget_or_compute("key", compute, (4,), {}))  # cached
...
>>> asyncio.run(main())
8
8
>>> calls
[4]

persista.cache.Cache.amemoize

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

Decorate a function so its return values are cached.

This is the async counterpart of :meth:memoize, for use with an async backing store. Works on both sync and async functions (async def); the wrapped function is always a coroutine function, since the backing store is only accessible through await.

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. See :meth:aset.	_UNSET
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'json'
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.	False

Returns:

Type	Description
Callable[[Callable[..., T] | Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]	A decorator that wraps a sync or async function with
Callable[[Callable[..., T] | Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]	caching, always returning a coroutine function.

Raises:

Type	Description
ValueError	If ttl is negative.

Example

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

persista.cache.Cache.aopen async

aopen() -> None

Async equivalent of :meth:open.

persista.cache.Cache.aset async

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

Add a value to the cache.

This is the async counterpart of :meth:set, for use with an async backing store.

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 when not given. None means the entry never expires. 0 means the value is not written to the store at all, evicting any existing entry for key instead. Must be non-negative.	_UNSET

Raises:

Type	Description
ValueError	If ttl is negative.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("greeting", "hello")
...         print(await cache.aget("greeting"))
...
>>> asyncio.run(main())
hello

persista.cache.Cache.aset_many async

aset_many(
    items: dict[str, Any], ttl: float | None = _UNSET
) -> None

Add multiple values in a single batched store write.

This is the async counterpart of :meth:set_many, for use with an async backing store.

Parameters:

Name	Type	Description	Default
items	dict[str, Any]	A dict mapping each key to the value to cache under it. Values 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 entries expire. See :meth:aset.	_UNSET

Raises:

Type	Description
ValueError	If ttl is negative.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset_many({"a": "hello", "b": "world"})
...         print(sorted((await cache.aget_many(["a", "b"])).items()))
...
>>> asyncio.run(main())
[('a', 'hello'), ('b', 'world')]

persista.cache.Cache.atry_get async

atry_get(key: str) -> tuple[bool, Any]

Look up a key, returning both hit/miss and the value.

This is the async counterpart of :meth:try_get.

Parameters:

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

Returns:

Type	Description
bool	A (hit, value) tuple. hit is True only when
Any	key exists in the store and has not expired.
tuple[bool, Any]	value is None when hit is False.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("key", None)
...         print(await cache.atry_get("key"))
...         print(await cache.atry_get("missing"))
...
>>> asyncio.run(main())
(True, None)
(False, None)

persista.cache.Cache.atry_get_many async

atry_get_many(keys: list[str]) -> dict[str, Any]

Look up multiple keys in a single batched store lookup.

This is the async counterpart of :meth:try_get_many, for use with an async backing store.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to look up.	required

Returns:

Name	Type	Description
	dict[str, Any]	A dict mapping each key that is a hit to its cached value.
See	dict[str, Any]	meth:try_get_many for the exact hit/miss semantics.

Example

>>> import asyncio
>>> from persista.cache.cache import Cache
>>> async def main():
...     async with Cache() as cache:
...         await cache.aset("a", "hello")
...         await cache.aset("b", "world")
...         print(sorted((await cache.atry_get_many(["a", "b", "missing"])).items()))
...
>>> asyncio.run(main())
[('a', 'hello'), ('b', 'world')]

persista.cache.Cache.clear

clear() -> None

Remove every entry from the cache, expired or not.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("greeting", "hello")
...     cache.clear()
...     cache.get("greeting") is None
...
True

persista.cache.Cache.close

close() -> None

Close the cache's backing store and release any underlying resources.

Repeated calls are safe (idempotent), since the underlying :meth:BaseStore.close is idempotent.

persista.cache.Cache.contains

contains(key: str) -> bool

Indicate whether a key is present and unexpired.

Parameters:

Name	Type	Description	Default
key	str	The key to check.	required

Returns:

Type	Description
bool	True if key has an entry in the cache that has not
bool	expired, otherwise False. If the entry has expired,
bool	it is evicted from the backing store as a side effect of
bool	this call, as in :meth:get.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("greeting", "hello")
...     cache.contains("greeting")
...     cache.contains("missing")
...
True
False

persista.cache.Cache.contains_many

contains_many(keys: list[str]) -> list[bool]

Check presence of multiple keys in a single batched store lookup.

Unlike calling :meth:contains once per key, this issues one self._store.get_many call for the whole batch, which matters for stores where each lookup is a network round trip (e.g. Redis, Postgres).

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to check.	required

Returns:

Type	Description
list[bool]	A list of booleans, in the same order as keys, where
list[bool]	each entry is True if the corresponding key is a hit
list[bool]	-- present and unexpired -- and False otherwise.
list[bool]	Expired entries are evicted from the backing store as a
list[bool]	side effect of this call, as in :meth:get_many.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("a", "hello")
...     cache.contains_many(["a", "b"])
...
[True, False]

persista.cache.Cache.delete

delete(key: str) -> None

Remove a single entry from the cache, if present.

Unlike :meth:set with ttl=0, this does not require a value to be given.

Parameters:

Name	Type	Description	Default
key	str	The key to remove.	required

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("greeting", "hello")
...     cache.delete("greeting")
...     cache.get("greeting") is None
...
True

persista.cache.Cache.delete_many

delete_many(keys: list[str]) -> None

Remove multiple entries from the cache in a single batched store write, if present.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to remove.	required

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set_many({"a": "hello", "b": "world"})
...     cache.delete_many(["a", "b"])
...     cache.try_get_many(["a", "b"])
...
{}

persista.cache.Cache.get

get(key: str, default: Any = None) -> Any

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 default is returned.

Parameters:

Name	Type	Description	Default
key	str	The key to look up.	required
default	Any	The value to return if the key is missing or its entry has expired. Pass the :data:~persista.cache.cache.MISSING sentinel here to distinguish a cache miss from a cached None.	None

Returns:

Type	Description
Any	The cached value, or default on a miss.

Example

>>> from persista.cache.cache import MISSING, Cache
>>> with Cache() as cache:
...     cache.set("greeting", "hello")
...     cache.get("greeting")
...     cache.get("missing") is None
...     cache.set("empty", None)
...     cache.get("empty", MISSING) is MISSING
...     cache.get("missing", MISSING) is MISSING
...
'hello'
True
False
True

persista.cache.Cache.get_many

get_many(
    keys: list[str], default: Any = None
) -> dict[str, Any]

Retrieve multiple values in a single batched store lookup.

Unlike calling :meth:get (or :meth:contains followed by :meth:get) once per key, this issues one self._store.get_many call for the whole batch, which matters for stores where each lookup is a network round trip (e.g. Redis, Postgres).

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to look up.	required
default	Any	The value to map a key to if it is missing or its entry has expired. Pass the :data:~persista.cache.cache.MISSING sentinel here to distinguish a cache miss from a cached None.	None

Returns:

Type	Description
dict[str, Any]	A dict mapping every key in keys to its cached value
dict[str, Any]	on a hit, or to default on a miss. Expired entries are
dict[str, Any]	evicted from the backing store as a side effect of this
dict[str, Any]	call, as in :meth:get.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("a", "hello")
...     cache.set("b", "world")
...     sorted(cache.get_many(["a", "b", "missing"]).items())
...
[('a', 'hello'), ('b', 'world'), ('missing', None)]

persista.cache.Cache.get_or_compute

get_or_compute(
    key: str,
    fn: Callable[..., T],
    args: tuple[Any, ...],
    kwargs: dict[str, Any],
    ttl: float | None = _UNSET,
) -> T

Return the cached value for key, computing and storing it on a cache miss.

Parameters:

Name	Type	Description	Default
key	str	The key to look up and, on a miss, store the result under.	required
fn	Callable[..., T]	The function to call to compute the value when key is not in the cache.	required
args	tuple[Any, ...]	Positional arguments passed to fn on a miss.	required
kwargs	dict[str, Any]	Keyword arguments passed to fn on a miss.	required
ttl	float | None	The time-to-live, in seconds, applied when storing a freshly computed value. See :meth:set.	_UNSET

Returns:

Type	Description
T	The cached value on a hit, otherwise the value returned by
T	fn(*args, **kwargs).

Example

>>> from persista.cache.cache import Cache
>>> calls = []
>>> def compute(x):
...     calls.append(x)
...     return x * 2
...
>>> with Cache() as cache:
...     cache.get_or_compute("key", compute, (4,), {})
...     cache.get_or_compute("key", compute, (4,), {})  # served from the cache
...
8
8
>>> calls
[4]

persista.cache.Cache.memoize

memoize(
    ttl: float | None = _UNSET,
    strategy: str = "json",
    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. See :meth:set.	_UNSET
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'json'
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.	False

Returns:

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

Raises:

Type	Description
ValueError	If ttl is negative.

Example

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

persista.cache.Cache.open

open() -> None

Connect the backing store and prepare the cache for use.

Repeated calls are safe (idempotent), since the underlying :meth:BaseStore.open is idempotent.

persista.cache.Cache.set

set(
    key: str, value: Any, ttl: float | None = _UNSET
) -> 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 when not given. None means the entry never expires. 0 means the value is not written to the store at all, evicting any existing entry for key instead. Must be non-negative.	_UNSET

Raises:

Type	Description
ValueError	If ttl is negative.

Example

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

persista.cache.Cache.set_many

set_many(
    items: dict[str, Any], ttl: float | None = _UNSET
) -> None

Add multiple values in a single batched store write.

Unlike calling :meth:set once per item, this issues one self._store.set_many call for the whole batch, which matters for stores where each write is a network round trip (e.g. Redis, Postgres). A single ttl applies to every item in the batch.

Parameters:

Name	Type	Description	Default
items	dict[str, Any]	A dict mapping each key to the value to cache under it. Values 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 entries expire. See :meth:set.	_UNSET

Raises:

Type	Description
ValueError	If ttl is negative.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set_many({"a": "hello", "b": "world"})
...     sorted(cache.get_many(["a", "b"]).items())
...
[('a', 'hello'), ('b', 'world')]

persista.cache.Cache.try_get

try_get(key: str) -> tuple[bool, Any]

Look up a key, returning both hit/miss and the value.

Unlike :meth:get, this distinguishes a cache miss from a cached None without needing the :data:MISSING sentinel.

Parameters:

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

Returns:

Type	Description
bool	A (hit, value) tuple. hit is True only when
Any	key exists in the store and has not expired.
tuple[bool, Any]	value is None when hit is False.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("key", None)
...     cache.try_get("key")
...     cache.try_get("missing")
...
(True, None)
(False, None)

persista.cache.Cache.try_get_many

try_get_many(keys: list[str]) -> dict[str, Any]

Look up multiple keys in a single batched store lookup.

Unlike :meth:get_many, this distinguishes a cache miss from a cached None without needing the :data:MISSING sentinel, by omitting missed keys entirely.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to look up.	required

Returns:

Type	Description
dict[str, Any]	A dict mapping each key that is a hit -- present and
dict[str, Any]	unexpired -- to its cached value. Keys that are missing
dict[str, Any]	or expired are omitted entirely rather than mapped to
dict[str, Any]	None, so a hit can always be distinguished from a
dict[str, Any]	miss with in. Expired entries are evicted from the
dict[str, Any]	backing store as a side effect of this call, as in
dict[str, Any]	meth:try_get.

Example

>>> from persista.cache.cache import Cache
>>> with Cache() as cache:
...     cache.set("a", "hello")
...     cache.set("b", "world")
...     sorted(cache.try_get_many(["a", "b", "missing"]).items())
...
[('a', 'hello'), ('b', 'world')]

persista.cache.async_cached

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

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

Looks up :func:get_cache on every call, so replacing the shared cache via :func:set_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	float | None	The time-to-live, in seconds, applied to cached results. Defaults to the cache's default_ttl when not given. See :meth:~persista.cache.cache.Cache.aset.	_UNSET
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'json'
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 negative.

Example

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

persista.cache.cached

cached(
    ttl: float | None = _UNSET,
    strategy: str = "json",
    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_cache on every call, so replacing the shared cache via :func:set_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	float | None	The time-to-live, in seconds, applied to cached results. Defaults to the cache's default_ttl when not given. See :meth:~persista.cache.cache.Cache.set.	_UNSET
strategy	str	The serialization strategy used to compute the cache key. Either "json" or "pickle". See :func:~persista.cache.utils.make_key.	'json'
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 negative.

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_cache

get_cache() -> Cache

Return the shared default cache.

Returns:

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

Example

>>> from persista.cache.interface import get_cache
>>> cache = get_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 = "json",
    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).	'json'
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.resolve_cache

resolve_cache(cache: Cache | dict[str, Any]) -> Cache

Resolve a :class:~persista.cache.Cache instance from an existing object or a configuration dictionary.

If cache is already a :class:~persista.cache.Cache instance it is returned as-is. If it is a :class:dict, it is treated as an objectory factory configuration and instantiated via :func:objectory.factory. See :func:~coola.factory.resolve_object for details.

Parameters:

Name	Type	Description	Default
cache	Cache | dict[str, Any]	Either a fully configured :class:~persista.cache.Cache instance, or a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name).	required

Returns:

Type	Description
Cache	A configured :class:~persista.cache.Cache instance.

Raises:

Type	Description
TypeError	If the resolved object is not a :class:~persista.cache.Cache instance.

Example

>>> from persista.cache import Cache, resolve_cache
>>> # From an existing instance:
>>> cache = resolve_cache(Cache())
>>> # From a configuration dictionary:
>>> cache = resolve_cache({"_target_": "persista.cache.Cache"})

persista.cache.set_cache

set_cache(cache: Cache) -> None

Replace the shared default cache.

Parameters:

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

Example

>>> from persista.cache import Cache
>>> from persista.cache import get_cache, set_cache
>>> previous = get_cache()
>>> new_cache = Cache(default_ttl=60)
>>> new_cache.open()
>>> set_cache(new_cache)
>>> get_cache().default_ttl
60
>>> set_cache(previous)  # restore the previous default cache

persista.cache.split_get_many

split_get_many(
    keys: list[str], values: dict[str, Any], default: Any
) -> tuple[list[str], list[str]]

Split keys into present and missing lists based on the output of :meth:~persista.cache.cache.Cache.get_many.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to split.	required
values	dict[str, Any]	The dict returned by Cache.get_many, mapping every key in keys to its cached value or to default.	required
default	Any	The same default value that was passed to Cache.get_many, used to detect misses. Pass the :data:~persista.cache.cache.MISSING sentinel here if that is what was passed to get_many, so that a cached None is not mistaken for a miss.	required

Returns:

Type	Description
list[str]	A (present, missing) tuple, each a list of keys in the
list[str]	same relative order as keys.

Example

>>> from persista.cache.utils import split_get_many
>>> split_get_many(["a", "b", "c"], {"a": 1, "b": None, "c": 3}, None)
(['a', 'c'], ['b'])

persista.cache.split_try_get_many

split_try_get_many(
    keys: list[str], values: dict[str, Any]
) -> tuple[list[str], list[str]]

Split keys into present and missing lists based on the output of :meth:~persista.cache.cache.Cache.try_get_many.

Parameters:

Name	Type	Description	Default
keys	list[str]	The keys to split.	required
values	dict[str, Any]	The dict returned by Cache.try_get_many, mapping each hit key to its cached value and omitting missed keys entirely.	required

Returns:

Type	Description
list[str]	A (present, missing) tuple, each a list of keys in the
list[str]	same relative order as keys.

Example

>>> from persista.cache.utils import split_try_get_many
>>> split_try_get_many(["a", "b", "c"], {"a": 1, "c": 3})
(['a', 'c'], ['b'])

persista.cache.factory

Contain factories for caches.

persista.cache.factory.BaseCacheFactory

Bases: ABC

Abstract base class for :class:~persista.cache.Cache factories.

Subclasses implement :meth:make_cache to instantiate and return a configured :class:~persista.cache.Cache object. This pattern decouples cache creation from the rest of the codebase, making it easy to swap how a cache is built (e.g. a shared instance vs. a fresh one per call) without changing call sites.

Example

>>> from persista.cache import Cache
>>> from persista.cache.factory import BaseCacheFactory
>>> class MyCacheFactory(BaseCacheFactory):
...     def make_cache(self) -> Cache:
...         return Cache()
...
>>> factory = MyCacheFactory()
>>> cache = factory.make_cache()

persista.cache.factory.BaseCacheFactory.make_cache abstractmethod

make_cache() -> Cache

Create and return a configured Cache instance.

Returns:

Name	Type	Description
A	Cache	class:~persista.cache.Cache
	Cache	instance ready for use.

persista.cache.factory.CacheFactory

Bases: BaseCacheFactory, MultilineDisplayMixin

A concrete Cache factory that wraps a pre-built :class:~persista.cache.Cache instance.

Use this when the cache is already instantiated and you simply want to wrap it in the :class:~BaseCacheFactory interface — for example, when injecting a fixed cache into a component that expects a factory.

Parameters:

Name	Type	Description	Default
cache	Cache	A fully configured :class:~persista.cache.Cache instance to return from :meth:make_cache.	required

Example

>>> from persista.cache import Cache
>>> from persista.cache.factory import CacheFactory
>>> factory = CacheFactory(Cache())
>>> cache = factory.make_cache()

persista.cache.factory.ConfigurableCacheFactory

Bases: BaseCacheFactory, MultilineDisplayMixin

A concrete Cache factory that accepts either a pre-built :class:~persista.cache.Cache instance or a configuration dictionary.

When a dict is provided it is resolved at each :meth:make_cache call via :func:~persista.cache.resolve.resolve_cache, which uses objectory to instantiate the configured class. When an instance is provided it is returned as-is.

Parameters:

Name	Type	Description	Default
cache	Cache | dict[str, Any]	A fully configured :class:~persista.cache.Cache instance, or a :class:dict containing an objectory factory specification (must include a "_target_" key pointing to the fully-qualified class name).	required

Example

>>> from persista.cache import Cache
>>> from persista.cache.factory import ConfigurableCacheFactory
>>> factory = ConfigurableCacheFactory(Cache())
>>> cache = factory.make_cache()

persista.cache.factory.StoreCacheFactory

Bases: BaseCacheFactory, MultilineDisplayMixin

A concrete Cache factory that builds its backing store from a :class:~persista.store.factory.BaseStoreFactory.

Use this when the store itself needs to be freshly created (e.g. a new connection, a new in-memory dict) each time a :class:~persista.cache.Cache is requested, rather than sharing one store instance across every cache.

Parameters:

Name	Type	Description	Default
store_factory	BaseStoreFactory	The factory used to create the backing store passed to each :class:~persista.cache.Cache built by :meth:make_cache.	required
default_ttl	float | None	The default time-to-live, in seconds, forwarded to each created :class:~persista.cache.Cache. See :class:~persista.cache.Cache for details.	None

Example

>>> from persista.cache.factory import StoreCacheFactory
>>> from persista.store.factory import StoreFactory
>>> from persista.store import InMemoryStore
>>> factory = StoreCacheFactory(StoreFactory(InMemoryStore()))
>>> with factory.make_cache() as cache:
...     cache.set("greeting", "hello")
...     cache.get("greeting")
...
'hello'