Skip to content

cache

CacheItem

Simple cache item with lifetime and default factory for default value.

Default factory should return default value that is used on init and on reset.

Parameters:

Name Type Description Default
default_factory Optional[callable]

Function that returns default value used on init and on reset.

None
lifetime Optional[int]

Lifetime of the cache data in seconds. Default lifetime is 120 seconds.

None
Source code in client/ayon_core/lib/cache.py
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
class CacheItem:
    """Simple cache item with lifetime and default factory for default value.

    Default factory should return default value that is used on init
        and on reset.

    Args:
        default_factory (Optional[callable]): Function that returns default
            value used on init and on reset.
        lifetime (Optional[int]): Lifetime of the cache data in seconds.
            Default lifetime is 120 seconds.

    """
    def __init__(self, default_factory=None, lifetime=None):
        if lifetime is None:
            lifetime = 120
        self._lifetime = lifetime
        self._last_update = None
        if default_factory is None:
            default_factory = _default_factory_func
        self._default_factory = default_factory
        self._data = default_factory()

    @property
    def is_valid(self):
        """Is cache valid to use.

        Return:
            bool: True if cache is valid, False otherwise.

        """
        if self._last_update is None:
            return False

        return (time.time() - self._last_update) < self._lifetime

    def set_lifetime(self, lifetime):
        """Change lifetime of cache item.

        Args:
            lifetime (int): Lifetime of the cache data in seconds.
        """

        self._lifetime = lifetime

    def set_invalid(self):
        """Set cache as invalid."""

        self._last_update = None

    def reset(self):
        """Set cache as invalid and reset data."""

        self._last_update = None
        self._data = self._default_factory()

    def get_data(self):
        """Receive cached data.

        Returns:
            Any: Any data that are cached.

        """
        return self._data

    def update_data(self, data):
        """Update cache data.

        Args:
            data (Any): Any data that are cached.

        """
        self._data = data
        self._last_update = time.time()

is_valid property

Is cache valid to use.

Return

bool: True if cache is valid, False otherwise.

get_data()

Receive cached data.

Returns:

Name Type Description
Any

Any data that are cached.

Source code in client/ayon_core/lib/cache.py
70
71
72
73
74
75
76
77
def get_data(self):
    """Receive cached data.

    Returns:
        Any: Any data that are cached.

    """
    return self._data

reset()

Set cache as invalid and reset data.

Source code in client/ayon_core/lib/cache.py
64
65
66
67
68
def reset(self):
    """Set cache as invalid and reset data."""

    self._last_update = None
    self._data = self._default_factory()

set_invalid()

Set cache as invalid.

Source code in client/ayon_core/lib/cache.py
59
60
61
62
def set_invalid(self):
    """Set cache as invalid."""

    self._last_update = None

set_lifetime(lifetime)

Change lifetime of cache item.

Parameters:

Name Type Description Default
lifetime int

Lifetime of the cache data in seconds.

required
Source code in client/ayon_core/lib/cache.py
50
51
52
53
54
55
56
57
def set_lifetime(self, lifetime):
    """Change lifetime of cache item.

    Args:
        lifetime (int): Lifetime of the cache data in seconds.
    """

    self._lifetime = lifetime

update_data(data)

Update cache data.

Parameters:

Name Type Description Default
data Any

Any data that are cached.

required
Source code in client/ayon_core/lib/cache.py
79
80
81
82
83
84
85
86
87
def update_data(self, data):
    """Update cache data.

    Args:
        data (Any): Any data that are cached.

    """
    self._data = data
    self._last_update = time.time()

NestedCacheItem

Helper for cached items stored in nested structure.

Example

cache = NestedCacheItem(levels=2, default_factory=lambda: 0) cache["a"]["b"].is_valid False cache["a"]["b"].get_data() 0 cache["a"]["b"] = 1 cache["a"]["b"].is_valid True cache["a"]["b"].get_data() 1 cache.reset() cache["a"]["b"].is_valid False

Parameters:

Name Type Description Default
levels int

Number of nested levels where read cache is stored.

1
default_factory Optional[callable]

Function that returns default value used on init and on reset.

None
lifetime Optional[int]

Lifetime of the cache data in seconds. Default value is based on default value of 'CacheItem'.

None
_init_info Optional[InitInfo]

Private argument. Init info for nested cache where created from parent item.

None
Source code in client/ayon_core/lib/cache.py
 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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
class NestedCacheItem:
    """Helper for cached items stored in nested structure.

    Example:
        >>> cache = NestedCacheItem(levels=2, default_factory=lambda: 0)
        >>> cache["a"]["b"].is_valid
        False
        >>> cache["a"]["b"].get_data()
        0
        >>> cache["a"]["b"] = 1
        >>> cache["a"]["b"].is_valid
        True
        >>> cache["a"]["b"].get_data()
        1
        >>> cache.reset()
        >>> cache["a"]["b"].is_valid
        False

    Args:
        levels (int): Number of nested levels where read cache is stored.
        default_factory (Optional[callable]): Function that returns default
            value used on init and on reset.
        lifetime (Optional[int]): Lifetime of the cache data in seconds.
            Default value is based on default value of 'CacheItem'.
        _init_info (Optional[InitInfo]): Private argument. Init info for
            nested cache where created from parent item.

    """
    def __init__(
        self, levels=1, default_factory=None, lifetime=None, _init_info=None
    ):
        if levels < 1:
            raise ValueError("Nested levels must be greater than 0")
        self._data_by_key = {}
        if _init_info is None:
            _init_info = InitInfo(default_factory, lifetime)
        self._init_info = _init_info
        self._levels = levels

    def __getitem__(self, key):
        """Get cached data.

        Args:
            key (str): Key of the cache item.

        Returns:
            Union[NestedCacheItem, CacheItem]: Cache item.

        """
        cache = self._data_by_key.get(key)
        if cache is None:
            if self._levels > 1:
                cache = NestedCacheItem(
                    levels=self._levels - 1,
                    _init_info=self._init_info
                )
            else:
                cache = CacheItem(
                    self._init_info.default_factory,
                    self._init_info.lifetime
                )
            self._data_by_key[key] = cache
        return cache

    def __setitem__(self, key, value):
        """Update cached data.

        Args:
            key (str): Key of the cache item.
            value (Any): Any data that are cached.

        """
        if self._levels > 1:
            raise AttributeError((
                "{} does not support '__setitem__'. Lower nested level by {}"
            ).format(self.__class__.__name__, self._levels - 1))
        cache = self[key]
        cache.update_data(value)

    def get(self, key):
        """Get cached data.

        Args:
            key (str): Key of the cache item.

        Returns:
            Union[NestedCacheItem, CacheItem]: Cache item.

        """
        return self[key]

    def cached_count(self):
        """Amount of cached items.

        Returns:
            int: Amount of cached items.

        """
        return len(self._data_by_key)

    def clear_key(self, key):
        """Clear cached item by key.

        Args:
            key (str): Key of the cache item.

        """
        self._data_by_key.pop(key, None)

    def clear_invalid(self):
        """Clear all invalid cache items.

        Note:
            To clear all cache items use 'reset'.

        """
        changed = {}
        children_are_nested = self._levels > 1
        for key, cache in tuple(self._data_by_key.items()):
            if children_are_nested:
                output = cache.clear_invalid()
                if output:
                    changed[key] = output
                if not cache.cached_count():
                    self._data_by_key.pop(key)
            elif not cache.is_valid:
                changed[key] = cache.get_data()
                self._data_by_key.pop(key)
        return changed

    def reset(self):
        """Reset cache.

        Note:
            To clear only invalid cache items use 'clear_invalid'.

        """
        self._data_by_key = {}

    def set_lifetime(self, lifetime):
        """Change lifetime of all children cache items.

        Args:
            lifetime (int): Lifetime of the cache data in seconds.

        """
        self._init_info.lifetime = lifetime
        for cache in self._data_by_key.values():
            cache.set_lifetime(lifetime)

    @property
    def is_valid(self):
        """Raise reasonable error when called on wrong level.

        Raises:
            AttributeError: If called on nested cache item.

        """
        raise AttributeError((
            "{} does not support 'is_valid'. Lower nested level by '{}'"
        ).format(self.__class__.__name__, self._levels))

is_valid property

Raise reasonable error when called on wrong level.

Raises:

Type Description
AttributeError

If called on nested cache item.

__getitem__(key)

Get cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required

Returns:

Type Description

Union[NestedCacheItem, CacheItem]: Cache item.

Source code in client/ayon_core/lib/cache.py
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def __getitem__(self, key):
    """Get cached data.

    Args:
        key (str): Key of the cache item.

    Returns:
        Union[NestedCacheItem, CacheItem]: Cache item.

    """
    cache = self._data_by_key.get(key)
    if cache is None:
        if self._levels > 1:
            cache = NestedCacheItem(
                levels=self._levels - 1,
                _init_info=self._init_info
            )
        else:
            cache = CacheItem(
                self._init_info.default_factory,
                self._init_info.lifetime
            )
        self._data_by_key[key] = cache
    return cache

__setitem__(key, value)

Update cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required
value Any

Any data that are cached.

required
Source code in client/ayon_core/lib/cache.py
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def __setitem__(self, key, value):
    """Update cached data.

    Args:
        key (str): Key of the cache item.
        value (Any): Any data that are cached.

    """
    if self._levels > 1:
        raise AttributeError((
            "{} does not support '__setitem__'. Lower nested level by {}"
        ).format(self.__class__.__name__, self._levels - 1))
    cache = self[key]
    cache.update_data(value)

cached_count()

Amount of cached items.

Returns:

Name Type Description
int

Amount of cached items.

Source code in client/ayon_core/lib/cache.py
181
182
183
184
185
186
187
188
def cached_count(self):
    """Amount of cached items.

    Returns:
        int: Amount of cached items.

    """
    return len(self._data_by_key)

clear_invalid()

Clear all invalid cache items.

Note

To clear all cache items use 'reset'.

Source code in client/ayon_core/lib/cache.py
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def clear_invalid(self):
    """Clear all invalid cache items.

    Note:
        To clear all cache items use 'reset'.

    """
    changed = {}
    children_are_nested = self._levels > 1
    for key, cache in tuple(self._data_by_key.items()):
        if children_are_nested:
            output = cache.clear_invalid()
            if output:
                changed[key] = output
            if not cache.cached_count():
                self._data_by_key.pop(key)
        elif not cache.is_valid:
            changed[key] = cache.get_data()
            self._data_by_key.pop(key)
    return changed

clear_key(key)

Clear cached item by key.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required
Source code in client/ayon_core/lib/cache.py
190
191
192
193
194
195
196
197
def clear_key(self, key):
    """Clear cached item by key.

    Args:
        key (str): Key of the cache item.

    """
    self._data_by_key.pop(key, None)

get(key)

Get cached data.

Parameters:

Name Type Description Default
key str

Key of the cache item.

required

Returns:

Type Description

Union[NestedCacheItem, CacheItem]: Cache item.

Source code in client/ayon_core/lib/cache.py
169
170
171
172
173
174
175
176
177
178
179
def get(self, key):
    """Get cached data.

    Args:
        key (str): Key of the cache item.

    Returns:
        Union[NestedCacheItem, CacheItem]: Cache item.

    """
    return self[key]

reset()

Reset cache.

Note

To clear only invalid cache items use 'clear_invalid'.

Source code in client/ayon_core/lib/cache.py
220
221
222
223
224
225
226
227
def reset(self):
    """Reset cache.

    Note:
        To clear only invalid cache items use 'clear_invalid'.

    """
    self._data_by_key = {}

set_lifetime(lifetime)

Change lifetime of all children cache items.

Parameters:

Name Type Description Default
lifetime int

Lifetime of the cache data in seconds.

required
Source code in client/ayon_core/lib/cache.py
229
230
231
232
233
234
235
236
237
238
def set_lifetime(self, lifetime):
    """Change lifetime of all children cache items.

    Args:
        lifetime (int): Lifetime of the cache data in seconds.

    """
    self._init_info.lifetime = lifetime
    for cache in self._data_by_key.values():
        cache.set_lifetime(lifetime)