Caches

Synchronous cache

Synchronous cache.

Source code in cachetory/caches/sync.py

class Cache(
    AbstractContextManager,  # type: ignore[type-arg]
    Generic[ValueT, WireT],
):
    """Synchronous cache."""

    __slots__ = ("_serializer", "_backend", "_prefix")

    def __init__(self, *, serializer: Serializer[ValueT, WireT], backend: SyncBackend[WireT], prefix: str = "") -> None:
        """
        Instantiate a cache.

        Args:
            prefix: backend key prefix
        """
        self._serializer = serializer
        self._backend = backend
        self._prefix = prefix

    def __getitem__(self, key: str) -> ValueT:
        """
        Retrieve a single value from the cache.

        Raises:
            KeyError: the key is not found in the cache

        Examples:
            >>> cache["key"] = 42
            >>>
            >>> assert cache["key"] == 42
            >>>
            >>> with pytest.raises(KeyError):
            >>>     _ = cache["missing"]
        """
        return self._serializer.deserialize(self._backend.get(f"{self._prefix}{key}"))

    def get(self, key: str, default: DefaultT = None) -> Union[ValueT, DefaultT]:  # type: ignore
        """
        Retrieve a single value from the cache.

        Args:
            key: cache key
            default: default value – if the key is not found

        Returns:
            Retrieved value if present, or `default` otherwise.

        Examples:
            >>> cache["key"] = 42
            >>> assert cache.get("key") == 42
            >>> assert cache.get("missing") is None
        """
        try:
            return self[key]
        except KeyError:
            return default

    def get_many(self, *keys: str) -> dict[str, ValueT]:
        """
        Retrieve many values from the cache.

        Returns:
            Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

        Examples:
            >>> cache["key"] = 42
            >>> assert cache.get_many("key", "missing") == {"key": 42}
        """
        return {
            f"{self._prefix}{key}": self._serializer.deserialize(data) for key, data in self._backend.get_many(*keys)
        }

    def expire_in(self, key: str, time_to_live: Optional[timedelta] = None) -> None:
        """
        Set the expiration time for the key.

        Args:
            key: cache key
            time_to_live: time to live, or `None` to make it eternal
        """
        return self._backend.expire_in(f"{self._prefix}{key}", time_to_live)

    def __setitem__(self, key: str, value: ValueT) -> None:
        """Set the cache item. To customize the behavior, use `set()`."""
        self._backend.set(f"{self._prefix}{key}", self._serializer.serialize(value), time_to_live=None)

    def set(  # noqa: A003
        self,
        key: str,
        value: ValueT,
        time_to_live: Optional[timedelta] = None,
        if_not_exists: bool = False,
    ) -> None:
        """
        Set the cache item.

        Args:
            key: cache key
            value: cached value
            time_to_live: time to live, or `None` for eternal caching
            if_not_exists: only set the item if it does not already exist
        """
        self._backend.set(
            f"{self._prefix}{key}",
            self._serializer.serialize(value),
            time_to_live=time_to_live,
            if_not_exists=if_not_exists,
        )

    def set_many(self, items: Union[Iterable[tuple[str, ValueT]], Mapping[str, ValueT]]) -> None:
        """
        Set many cache items at once.

        Examples:
            >>> cache.set_many({"foo": 42, "bar": 100500})
        """
        if isinstance(items, Mapping):
            items = items.items()
        self._backend.set_many((f"{self._prefix}{key}", self._serializer.serialize(value)) for key, value in items)

    def delete(self, key: str) -> bool:
        """
        Delete the cache item.

        Returns:
            `True` if the key has existed, `False` otherwise
        """
        return self._backend.delete(f"{self._prefix}{key}")

    def clear(self) -> None:
        """Delete all cache items."""
        return self._backend.clear()

    def __delitem__(self, key: str) -> None:
        """
        Delete the cache item.

        Raises:
            KeyError: the key didn't exist
        """
        if not self.delete(key):
            raise KeyError(key)

    def __exit__(
        self,
        exc_type: Optional[type[BaseException]],
        exc_val: Optional[BaseException],
        exc_tb: Optional[TracebackType],
    ) -> None:
        return self._backend.__exit__(exc_type, exc_val, exc_tb)

__init__

__init__(
    *,
    serializer: Serializer[ValueT, WireT],
    backend: SyncBackend[WireT],
    prefix: str = ""
) -> None

Instantiate a cache.

Parameters:

Name	Type	Description	Default
prefix	str	backend key prefix	''

Source code in cachetory/caches/sync.py

def __init__(self, *, serializer: Serializer[ValueT, WireT], backend: SyncBackend[WireT], prefix: str = "") -> None:
    """
    Instantiate a cache.

    Args:
        prefix: backend key prefix
    """
    self._serializer = serializer
    self._backend = backend
    self._prefix = prefix

__getitem__

__getitem__(key: str) -> ValueT

Retrieve a single value from the cache.

Raises:

Type	Description
KeyError	the key is not found in the cache

Examples:

>>> cache["key"] = 42
>>>
>>> assert cache["key"] == 42
>>>
>>> with pytest.raises(KeyError):
>>>     _ = cache["missing"]

Source code in cachetory/caches/sync.py

def __getitem__(self, key: str) -> ValueT:
    """
    Retrieve a single value from the cache.

    Raises:
        KeyError: the key is not found in the cache

    Examples:
        >>> cache["key"] = 42
        >>>
        >>> assert cache["key"] == 42
        >>>
        >>> with pytest.raises(KeyError):
        >>>     _ = cache["missing"]
    """
    return self._serializer.deserialize(self._backend.get(f"{self._prefix}{key}"))

get

get(key: str, default: DefaultT = None) -> Union[ValueT, DefaultT]

Retrieve a single value from the cache.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
default	DefaultT	default value – if the key is not found	None

Returns:

Type	Description
Union[ValueT, DefaultT]	Retrieved value if present, or default otherwise.

Examples:

>>> cache["key"] = 42
>>> assert cache.get("key") == 42
>>> assert cache.get("missing") is None

Source code in cachetory/caches/sync.py

def get(self, key: str, default: DefaultT = None) -> Union[ValueT, DefaultT]:  # type: ignore
    """
    Retrieve a single value from the cache.

    Args:
        key: cache key
        default: default value – if the key is not found

    Returns:
        Retrieved value if present, or `default` otherwise.

    Examples:
        >>> cache["key"] = 42
        >>> assert cache.get("key") == 42
        >>> assert cache.get("missing") is None
    """
    try:
        return self[key]
    except KeyError:
        return default

get_many

get_many(*keys: str) -> dict[str, ValueT]

Retrieve many values from the cache.

Returns:

Type	Description
dict[str, ValueT]	Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

Examples:

>>> cache["key"] = 42
>>> assert cache.get_many("key", "missing") == {"key": 42}

Source code in cachetory/caches/sync.py

def get_many(self, *keys: str) -> dict[str, ValueT]:
    """
    Retrieve many values from the cache.

    Returns:
        Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

    Examples:
        >>> cache["key"] = 42
        >>> assert cache.get_many("key", "missing") == {"key": 42}
    """
    return {
        f"{self._prefix}{key}": self._serializer.deserialize(data) for key, data in self._backend.get_many(*keys)
    }

expire_in

expire_in(key: str, time_to_live: Optional[timedelta] = None) -> None

Set the expiration time for the key.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
time_to_live	Optional[timedelta]	time to live, or None to make it eternal	None

Source code in cachetory/caches/sync.py

def expire_in(self, key: str, time_to_live: Optional[timedelta] = None) -> None:
    """
    Set the expiration time for the key.

    Args:
        key: cache key
        time_to_live: time to live, or `None` to make it eternal
    """
    return self._backend.expire_in(f"{self._prefix}{key}", time_to_live)

__setitem__

__setitem__(key: str, value: ValueT) -> None

Set the cache item. To customize the behavior, use set().

Source code in cachetory/caches/sync.py

def __setitem__(self, key: str, value: ValueT) -> None:
    """Set the cache item. To customize the behavior, use `set()`."""
    self._backend.set(f"{self._prefix}{key}", self._serializer.serialize(value), time_to_live=None)

set

set(
    key: str,
    value: ValueT,
    time_to_live: Optional[timedelta] = None,
    if_not_exists: bool = False,
) -> None

Set the cache item.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
value	ValueT	cached value	required
time_to_live	Optional[timedelta]	time to live, or None for eternal caching	None
if_not_exists	bool	only set the item if it does not already exist	False

Source code in cachetory/caches/sync.py

def set(  # noqa: A003
    self,
    key: str,
    value: ValueT,
    time_to_live: Optional[timedelta] = None,
    if_not_exists: bool = False,
) -> None:
    """
    Set the cache item.

    Args:
        key: cache key
        value: cached value
        time_to_live: time to live, or `None` for eternal caching
        if_not_exists: only set the item if it does not already exist
    """
    self._backend.set(
        f"{self._prefix}{key}",
        self._serializer.serialize(value),
        time_to_live=time_to_live,
        if_not_exists=if_not_exists,
    )

set_many

set_many(
    items: Union[Iterable[tuple[str, ValueT]], Mapping[str, ValueT]]
) -> None

Set many cache items at once.

Examples:

>>> cache.set_many({"foo": 42, "bar": 100500})

Source code in cachetory/caches/sync.py

def set_many(self, items: Union[Iterable[tuple[str, ValueT]], Mapping[str, ValueT]]) -> None:
    """
    Set many cache items at once.

    Examples:
        >>> cache.set_many({"foo": 42, "bar": 100500})
    """
    if isinstance(items, Mapping):
        items = items.items()
    self._backend.set_many((f"{self._prefix}{key}", self._serializer.serialize(value)) for key, value in items)

delete

delete(key: str) -> bool

Delete the cache item.

Returns:

Type	Description
bool	True if the key has existed, False otherwise

Source code in cachetory/caches/sync.py

def delete(self, key: str) -> bool:
    """
    Delete the cache item.

    Returns:
        `True` if the key has existed, `False` otherwise
    """
    return self._backend.delete(f"{self._prefix}{key}")

clear

clear() -> None

Delete all cache items.

Source code in cachetory/caches/sync.py

def clear(self) -> None:
    """Delete all cache items."""
    return self._backend.clear()

__delitem__

__delitem__(key: str) -> None

Delete the cache item.

Raises:

Type	Description
KeyError	the key didn't exist

Source code in cachetory/caches/sync.py

def __delitem__(self, key: str) -> None:
    """
    Delete the cache item.

    Raises:
        KeyError: the key didn't exist
    """
    if not self.delete(key):
        raise KeyError(key)

Asynchronous cache

Asynchronous cache.

Source code in cachetory/caches/async_.py

class Cache(
    AbstractAsyncContextManager,  # type: ignore[type-arg]
    Generic[ValueT, WireT],
):
    """Asynchronous cache."""

    __slots__ = ("_serializer", "_backend", "_serialize_executor", "_prefix")

    def __init__(
        self,
        *,
        serializer: Serializer[ValueT, WireT],
        backend: AsyncBackend[WireT],
        serialize_executor: Executor | NotSet | None = _NOT_SET,
        prefix: str = "",
    ) -> None:
        """
        Instantiate a cache.

        Args:
            serialize_executor:
                If specified, underlying serializing and deserializing will be performed
                using the executor (for example, `ProcessPoolExecutor`).
                This may be useful to better utilize CPU when caching large blobs.
                If not specified, (de)serialization is performed in the current thread.
            prefix: backend key prefix
        """
        self._serializer = serializer
        self._backend = backend
        self._serialize_executor = serialize_executor
        self._prefix = prefix

    async def get(self, key: str, default: DefaultT = None) -> ValueT | DefaultT:  # type: ignore
        """
        Retrieve a single value from the cache.

        Args:
            key: cache key
            default: default value – if the key is not found

        Returns:
            Retrieved value if present, or `default` otherwise.

        Examples:
            >>> await cache.set("key", 42)
            >>> assert await cache.get("key") == 42
            >>> assert await cache.get("missing") is None
        """
        try:
            data = await self._backend.get(f"{self._prefix}{key}")
        except KeyError:
            return default
        else:
            return await self._deserialize(data)

    async def get_many(self, *keys: str) -> dict[str, ValueT]:
        """
        Retrieve many values from the cache.

        Returns:
            Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

        Examples:
            >>> await memory_cache.set("foo", 42)
            >>> assert await memory_cache.get_many("foo", "bar") == {"foo": 42}
        """
        return {
            f"{self._prefix}{key}": await self._deserialize(data) async for key, data in self._backend.get_many(*keys)
        }

    async def expire_in(self, key: str, time_to_live: timedelta | None = None) -> None:
        """
        Set the expiration time for the key.

        Args:
            key: cache key
            time_to_live: time to live, or `None` to make it eternal
        """
        return await self._backend.expire_in(f"{self._prefix}{key}", time_to_live)

    async def set(  # noqa: A003
        self,
        key: str,
        value: ValueT,
        time_to_live: timedelta | None = None,
        if_not_exists: bool = False,
    ) -> None:
        """
        Set the cache item.

        Args:
            key: cache key
            value: cached value
            time_to_live: time to live, or `None` for eternal caching
            if_not_exists: only set the item if it does not already exist
        """
        await self._backend.set(
            f"{self._prefix}{key}",
            await self._serialize(value),
            time_to_live=time_to_live,
            if_not_exists=if_not_exists,
        )

    async def set_many(self, items: Iterable[tuple[str, ValueT]] | Mapping[str, ValueT]) -> None:
        """
        Set many cache items at once.

        Examples:
            >>> await cache.set_many({"foo": 42, "bar": 100500})
        """
        if isinstance(items, Mapping):
            items = items.items()
        await self._backend.set_many([(f"{self._prefix}{key}", await self._serialize(value)) for key, value in items])

    async def delete(self, key: str) -> bool:
        """
        Delete the cache item.

        Returns:
            `True` if the key has existed, `False` otherwise
        """
        return await self._backend.delete(f"{self._prefix}{key}")

    async def clear(self) -> None:
        """Delete all cache items."""
        return await self._backend.clear()

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        return await self._backend.__aexit__(exc_type, exc_value, traceback)

    async def _serialize(self, value: ValueT) -> WireT:
        if self._serialize_executor is _NOT_SET:
            return self._serializer.serialize(value)
        else:
            return await get_running_loop().run_in_executor(self._serialize_executor, self._serializer.serialize, value)

    async def _deserialize(self, data: WireT) -> ValueT:
        if self._serialize_executor is _NOT_SET:
            return self._serializer.deserialize(data)
        else:
            return await get_running_loop().run_in_executor(
                self._serialize_executor,
                self._serializer.deserialize,
                data,
            )

__init__

__init__(
    *,
    serializer: Serializer[ValueT, WireT],
    backend: AsyncBackend[WireT],
    serialize_executor: Executor | NotSet | None = _NOT_SET,
    prefix: str = ""
) -> None

Instantiate a cache.

Parameters:

Name	Type	Description	Default
serialize_executor	Executor | NotSet | None	If specified, underlying serializing and deserializing will be performed using the executor (for example, ProcessPoolExecutor). This may be useful to better utilize CPU when caching large blobs. If not specified, (de)serialization is performed in the current thread.	_NOT_SET
prefix	str	backend key prefix	''

Source code in cachetory/caches/async_.py

def __init__(
    self,
    *,
    serializer: Serializer[ValueT, WireT],
    backend: AsyncBackend[WireT],
    serialize_executor: Executor | NotSet | None = _NOT_SET,
    prefix: str = "",
) -> None:
    """
    Instantiate a cache.

    Args:
        serialize_executor:
            If specified, underlying serializing and deserializing will be performed
            using the executor (for example, `ProcessPoolExecutor`).
            This may be useful to better utilize CPU when caching large blobs.
            If not specified, (de)serialization is performed in the current thread.
        prefix: backend key prefix
    """
    self._serializer = serializer
    self._backend = backend
    self._serialize_executor = serialize_executor
    self._prefix = prefix

get async

get(key: str, default: DefaultT = None) -> ValueT | DefaultT

Retrieve a single value from the cache.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
default	DefaultT	default value – if the key is not found	None

Returns:

Type	Description
ValueT | DefaultT	Retrieved value if present, or default otherwise.

Examples:

>>> await cache.set("key", 42)
>>> assert await cache.get("key") == 42
>>> assert await cache.get("missing") is None

Source code in cachetory/caches/async_.py

async def get(self, key: str, default: DefaultT = None) -> ValueT | DefaultT:  # type: ignore
    """
    Retrieve a single value from the cache.

    Args:
        key: cache key
        default: default value – if the key is not found

    Returns:
        Retrieved value if present, or `default` otherwise.

    Examples:
        >>> await cache.set("key", 42)
        >>> assert await cache.get("key") == 42
        >>> assert await cache.get("missing") is None
    """
    try:
        data = await self._backend.get(f"{self._prefix}{key}")
    except KeyError:
        return default
    else:
        return await self._deserialize(data)

get_many async

get_many(*keys: str) -> dict[str, ValueT]

Retrieve many values from the cache.

Returns:

Type	Description
dict[str, ValueT]	Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

Examples:

>>> await memory_cache.set("foo", 42)
>>> assert await memory_cache.get_many("foo", "bar") == {"foo": 42}

Source code in cachetory/caches/async_.py

async def get_many(self, *keys: str) -> dict[str, ValueT]:
    """
    Retrieve many values from the cache.

    Returns:
        Dictionary of existing values indexed by their respective keys. Missing keys are omitted.

    Examples:
        >>> await memory_cache.set("foo", 42)
        >>> assert await memory_cache.get_many("foo", "bar") == {"foo": 42}
    """
    return {
        f"{self._prefix}{key}": await self._deserialize(data) async for key, data in self._backend.get_many(*keys)
    }

expire_in async

expire_in(key: str, time_to_live: timedelta | None = None) -> None

Set the expiration time for the key.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
time_to_live	timedelta | None	time to live, or None to make it eternal	None

Source code in cachetory/caches/async_.py

async def expire_in(self, key: str, time_to_live: timedelta | None = None) -> None:
    """
    Set the expiration time for the key.

    Args:
        key: cache key
        time_to_live: time to live, or `None` to make it eternal
    """
    return await self._backend.expire_in(f"{self._prefix}{key}", time_to_live)

set async

set(
    key: str,
    value: ValueT,
    time_to_live: timedelta | None = None,
    if_not_exists: bool = False,
) -> None

Set the cache item.

Parameters:

Name	Type	Description	Default
key	str	cache key	required
value	ValueT	cached value	required
time_to_live	timedelta | None	time to live, or None for eternal caching	None
if_not_exists	bool	only set the item if it does not already exist	False

Source code in cachetory/caches/async_.py

async def set(  # noqa: A003
    self,
    key: str,
    value: ValueT,
    time_to_live: timedelta | None = None,
    if_not_exists: bool = False,
) -> None:
    """
    Set the cache item.

    Args:
        key: cache key
        value: cached value
        time_to_live: time to live, or `None` for eternal caching
        if_not_exists: only set the item if it does not already exist
    """
    await self._backend.set(
        f"{self._prefix}{key}",
        await self._serialize(value),
        time_to_live=time_to_live,
        if_not_exists=if_not_exists,
    )

set_many async

set_many(items: Iterable[tuple[str, ValueT]] | Mapping[str, ValueT]) -> None

Set many cache items at once.

Examples:

>>> await cache.set_many({"foo": 42, "bar": 100500})

Source code in cachetory/caches/async_.py

async def set_many(self, items: Iterable[tuple[str, ValueT]] | Mapping[str, ValueT]) -> None:
    """
    Set many cache items at once.

    Examples:
        >>> await cache.set_many({"foo": 42, "bar": 100500})
    """
    if isinstance(items, Mapping):
        items = items.items()
    await self._backend.set_many([(f"{self._prefix}{key}", await self._serialize(value)) for key, value in items])

delete async

delete(key: str) -> bool

Delete the cache item.

Returns:

Type	Description
bool	True if the key has existed, False otherwise

Source code in cachetory/caches/async_.py

async def delete(self, key: str) -> bool:
    """
    Delete the cache item.

    Returns:
        `True` if the key has existed, `False` otherwise
    """
    return await self._backend.delete(f"{self._prefix}{key}")

clear async

clear() -> None

Delete all cache items.

Source code in cachetory/caches/async_.py

async def clear(self) -> None:
    """Delete all cache items."""
    return await self._backend.clear()