Coverage for picos/caching.py: 86.21%
87 statements
« prev ^ index » next coverage.py v6.5.0, created at 2023-03-26 07:46 +0000
« prev ^ index » next coverage.py v6.5.0, created at 2023-03-26 07:46 +0000
1# ------------------------------------------------------------------------------
2# Copyright (C) 2019 Maximilian Stahlberg
3#
4# This file is part of PICOS.
5#
6# PICOS is free software: you can redistribute it and/or modify it under the
7# terms of the GNU General Public License as published by the Free Software
8# Foundation, either version 3 of the License, or (at your option) any later
9# version.
10#
11# PICOS is distributed in the hope that it will be useful, but WITHOUT ANY
12# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
13# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License along with
16# this program. If not, see <http://www.gnu.org/licenses/>.
17# ------------------------------------------------------------------------------
19"""Caching helpers."""
21import functools
22from contextlib import contextmanager
24from .apidoc import api_end, api_start
26_API_START = api_start(globals())
27# -------------------------------
30#: The prefix used for storing cached values.
31CACHED_PREFIX = "_cached_"
33#: An attribute name whose presence unlocks the setter of cached properties.
34CACHED_PROP_UNLOCKED_TOKEN = "_CACHED_PROPERTIES_ARE_UNLOCKED"
37@contextmanager
38def unlocked_cached_properties(obj):
39 """Unlock the setters of cached instance attributes.
41 Normally, cached attributes are read-only properties. When the user first
42 reads them, the cache is populated with the value returned to the user, and
43 successive reads will return the cached value.
45 The user is allowed to empty the cache by using ``del`` on the variable, but
46 they may not assign a value to it. This context allows the programmer to
47 manually populate the cache by assigning a value to the property.
49 :Example:
51 >>> from picos.caching import cached_property, unlocked_cached_properties
52 >>> class A:
53 ... @cached_property
54 ... def p(self):
55 ... return 1
56 ...
57 >>> a = A()
58 >>> try:
59 ... a.p = 2
60 ... except AttributeError:
61 ... print("Not possible.")
62 ...
63 Not possible.
64 >>> with unlocked_cached_properties(a):
65 ... a.p = 2 # Populate the cache of a.p.
66 ...
67 >>> a.p
68 2
69 """
70 assert not hasattr(obj, CACHED_PROP_UNLOCKED_TOKEN)
71 setattr(obj, CACHED_PROP_UNLOCKED_TOKEN, None)
73 try:
74 yield
75 finally:
76 delattr(obj, CACHED_PROP_UNLOCKED_TOKEN)
79class cached_property(property):
80 """A read-only property whose result is cached."""
82 def __init__(self, fget=None, fset=None, fdel=None, doc=None): # noqa
83 if fget is None:
84 raise NotImplementedError("Unlike normal properties, cached "
85 "properties must be initialized with a getter.")
86 elif fget.__name__ == (lambda: None).__name__:
87 raise NotImplementedError("Cached properties cannot be used with a "
88 "lambda getter as lambdas do not have a unique name to identify"
89 " the cached value.")
91 if fset is not None:
92 raise AttributeError("Cannot have a custom setter on a cached "
93 "property as __set__ is reserved for cache population.")
95 if fdel is not None:
96 raise AttributeError("Cannot have a custom deleter on a cached "
97 "property as __delete__ is reserved for cache clearing.")
99 self._cache_name = CACHED_PREFIX + fget.__name__
101 property.__init__(self, fget, fset, fdel, doc)
103 self.__module__ = fget.__module__
105 def __get__(self, obj, objtype=None):
106 if obj is None:
107 return self
109 if self.fget is None:
110 raise AttributeError("unreadable attribute") # Mimic Python.
112 if not hasattr(obj, self._cache_name):
113 # Compute and cache the value.
114 setattr(obj, self._cache_name, self.fget(obj))
116 return getattr(obj, self._cache_name)
118 def __set__(self, obj, value):
119 if not hasattr(obj, CACHED_PROP_UNLOCKED_TOKEN):
120 raise AttributeError("can't set attribute") # Mimic Python.
122 # Populate the cache.
123 setattr(obj, self._cache_name, value)
125 def __delete__(self, obj):
126 # Empty the cache.
127 if hasattr(obj, self._cache_name):
128 delattr(obj, self._cache_name)
130 def getter(self, fget): # noqa
131 raise NotImplementedError(
132 "Cannot change the getter on a cached property.")
134 def setter(self, fset): # noqa
135 raise AttributeError("Cannot add a setter to a cached property.")
137 def deleter(self, fdel): # noqa
138 return AttributeError("Cannot add a deleter to a cached property.")
141class cached_selfinverse_property(cached_property):
142 """A read-only, self-inverse property whose result is cached."""
144 def __get__(self, obj, objtype=None):
145 if obj is None:
146 return self
148 if self.fget is None:
149 raise AttributeError("unreadable attribute") # Mimic Python.
151 if not hasattr(obj, self._cache_name):
152 # Compute and cache the value, populate the value's cache.
153 value = self.fget(obj)
154 setattr(obj, self._cache_name, value)
155 setattr(value, self._cache_name, obj)
157 return getattr(obj, self._cache_name)
160def cached_unary_operator(operator):
161 """Make a unary operator method cache its result.
163 This is supposed to be used for property-like special methods such as
164 ``__neg__`` where :func:`cached_property` can't be used.
165 """
166 cacheName = CACHED_PREFIX + operator.__name__
168 @functools.wraps(operator)
169 def wrapper(self):
170 if not hasattr(self, cacheName):
171 setattr(self, cacheName, operator(self))
172 return getattr(self, cacheName)
173 return wrapper
176def cached_selfinverse_unary_operator(operator):
177 """Make a self-inverse unary operator method cache its result.
179 This is supposed to be used for property-like special methods such as
180 ``__neg__`` where :func:`cached_property` can't be used.
182 .. warning::
183 The result returned by the wrapped operator must be a fresh object as
184 it will be modified.
185 """
186 cacheName = CACHED_PREFIX + operator.__name__
188 @functools.wraps(operator)
189 def wrapper(self):
190 if not hasattr(self, cacheName):
191 value = operator(self)
192 setattr(self, cacheName, value)
193 setattr(value, cacheName, self)
194 return getattr(self, cacheName)
195 return wrapper
198def empty_cache(obj):
199 """Clear all cached values of an object."""
200 for name in dir(obj):
201 if name.startswith(CACHED_PREFIX):
202 delattr(obj, name)
205def borrow_cache(target, source, names):
206 """Copy cached values from one object to another.
208 :param target: The object to populate the cache of.
209 :param source: The object to take cached values from.
210 :param names: Names of cached properties or functions to borrow.
211 """
212 for name in names:
213 cacheName = CACHED_PREFIX + name
214 if hasattr(source, cacheName):
215 setattr(target, cacheName, getattr(source, cacheName))
218# --------------------------------------
219__all__ = api_end(_API_START, globals())