Skip to content

Caches

Synchronous cache

Synchronous cache.

Source code in cachetory/caches/sync.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
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 __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 
20
21
22
23
24
25
26
27
28
29
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
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 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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 
84
85
86
87
88
89
90
91
92
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 
94
95
96
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 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
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 
121
122
123
124
125
126
127
128
129
130
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 
132
133
134
135
136
137
138
139
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}")

__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 
141
142
143
144
145
146
147
148
149
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 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
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 __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 
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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 
89
90
91
92
93
94
95
96
97
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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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 
122
123
124
125
126
127
128
129
130
131
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 
133
134
135
136
137
138
139
140
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}")