Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/wrapt/caching.py: 24%

1"""Caching decorators. Currently provides ``lru_cache``, a drop-in 

2replacement for ``functools.lru_cache`` with correct handling of instance 

3methods: a separate cache is maintained per instance, stored as an 

4attribute on the instance itself so it is cleaned up with the instance 

5by the garbage collector. For plain functions, class methods, and static 

6methods a single shared cache is used, matching ``functools.lru_cache``. 

7""" 

8 

9from functools import lru_cache as _functools_lru_cache 

10from functools import partial 

11 

12from .__wrapt__ import BoundFunctionWrapper, FunctionWrapper 

13from .decorators import decorator 

14from .synchronization import synchronized 

15 

16# Decorator that applies functools.lru_cache to the wrapped function. 

17# Unlike using functools.lru_cache directly, this works correctly with 

18# instance methods and class methods by maintaining a separate cache 

19# per instance or class. Each per-instance cache is stored as an 

20# attribute on the instance itself, so it is automatically cleaned up 

21# when the instance is garbage collected. For plain functions and 

22# static methods, a single cache is stored on the wrapper itself. 

23# The underlying caches are created lazily on first call, with 

24# double-checked locking via synchronized to avoid races in 

25# multi-threaded code. 

26# 

27# Custom FunctionWrapper and BoundFunctionWrapper subclasses are used 

28# so that cache_info() and cache_clear() are available directly on 

29# the decorated function. For bound methods, the BoundFunctionWrapper 

30# uses _self_instance to locate the per-instance cache. 

31 

32 

33class _BoundLRUCacheFunctionWrapper(BoundFunctionWrapper): 

34 

35 def _is_instance_method(self): 

36 return self._self_binding == "function" 

37 

38 def __call__(self, *args, **kwargs): 

39 parent = self._self_parent 

40 

41 # For class methods and static methods, use a single shared 

42 # cache stored on the parent wrapper. The cache is created 

43 # from the bound __wrapped__ since the parent's __wrapped__ 

44 # is the raw classmethod/staticmethod descriptor. 

45 

46 if not self._is_instance_method(): 

47 if parent._self_cache is None: 

48 with synchronized(parent): 

49 if parent._self_cache is None: 

50 parent._self_cache = _functools_lru_cache( 

51 **parent._self_lru_kwargs 

52 )(self.__wrapped__) 

53 

54 return parent._self_cache(*args, **kwargs) 

55 

56 # Instance method — per-instance cache stored as an attribute 

57 # on the instance so it is cleaned up with the instance by the 

58 # garbage collector. 

59 

60 instance = self._self_instance 

61 cache_attr = parent._self_cache_attr 

62 

63 cache = getattr(instance, cache_attr, None) 

64 

65 if cache is None: 

66 with synchronized(parent): 

67 cache = getattr(instance, cache_attr, None) 

68 

69 if cache is None: 

70 cache = _functools_lru_cache(**parent._self_lru_kwargs)( 

71 self.__wrapped__ 

72 ) 

73 

74 setattr(instance, cache_attr, cache) 

75 

76 return cache(*args, **kwargs) 

77 

78 def cache_info(self): 

79 """Return the cache statistics for this binding's cache, or 

80 ``None`` if the cache has not yet been created. 

81 """ 

82 

83 if not self._is_instance_method(): 

84 return self._self_parent.cache_info() 

85 

86 cache = getattr(self._self_instance, self._self_parent._self_cache_attr, None) 

87 

88 if cache is not None: 

89 return cache.cache_info() 

90 

91 return None 

92 

93 def cache_clear(self): 

94 """Clear this binding's cache.""" 

95 

96 if not self._is_instance_method(): 

97 self._self_parent.cache_clear() 

98 return 

99 

100 cache = getattr(self._self_instance, self._self_parent._self_cache_attr, None) 

101 

102 if cache is not None: 

103 cache.cache_clear() 

104 

105 def cache_parameters(self): 

106 """Return the parameters used to create the cache.""" 

107 

108 if not self._is_instance_method(): 

109 return self._self_parent.cache_parameters() 

110 

111 cache = getattr(self._self_instance, self._self_parent._self_cache_attr, None) 

112 

113 if cache is not None: 

114 return cache.cache_parameters() 

115 

116 return None 

117 

118 

119class _LRUCacheFunctionWrapper(FunctionWrapper): 

120 

121 __bound_function_wrapper__ = _BoundLRUCacheFunctionWrapper 

122 

123 def __init__(self, wrapped, wrapper, **kwargs): 

124 super().__init__(wrapped, wrapper, **kwargs) 

125 

126 # Extract the LRU cache configuration that was attached to the 

127 # wrapper function before it was passed to the decorator. 

128 

129 self._self_lru_kwargs = wrapper._self_lru_kwargs 

130 self._self_cache = None 

131 

132 # Use __func__ to get the name for classmethod/staticmethod 

133 # descriptors which lack __name__ on Python < 3.10. 

134 

135 name = getattr(wrapped, "__name__", None) 

136 

137 if name is None: 

138 name = wrapped.__func__.__name__ 

139 

140 self._self_cache_attr = "_lru_cache_" + name 

141 

142 def __call__(self, *args, **kwargs): 

143 # Plain function or static method — single cache stored 

144 # on the wrapper itself. 

145 

146 if self._self_cache is None: 

147 with synchronized(self): 

148 if self._self_cache is None: 

149 self._self_cache = _functools_lru_cache(**self._self_lru_kwargs)( 

150 self.__wrapped__ 

151 ) 

152 

153 return self._self_cache(*args, **kwargs) 

154 

155 def cache_info(self): 

156 """Return the cache statistics, or ``None`` if the cache has 

157 not yet been created. 

158 """ 

159 

160 if self._self_cache is not None: 

161 return self._self_cache.cache_info() 

162 

163 return None 

164 

165 def cache_clear(self): 

166 """Clear the cache and reset the statistics.""" 

167 

168 if self._self_cache is not None: 

169 self._self_cache.cache_clear() 

170 

171 def cache_parameters(self): 

172 """Return the parameters used to create the cache, or ``None`` 

173 if the cache has not yet been created. 

174 """ 

175 

176 if self._self_cache is not None: 

177 return self._self_cache.cache_parameters() 

178 

179 return None 

180 

181 

182def lru_cache(func=None, /, **kwargs): 

183 """A decorator that applies ``functools.lru_cache`` to the wrapped 

184 function, with correct handling for instance methods, class methods, 

185 and static methods. 

186 

187 For instance methods, a separate ``functools.lru_cache`` is 

188 maintained per instance. The cache is stored as an attribute on the 

189 instance itself, so it is automatically cleaned up when the instance 

190 is garbage collected. This means instances do not need to be 

191 hashable, each instance gets its own full ``maxsize`` budget, and 

192 no external mapping prevents garbage collection. 

193 

194 For plain functions, class methods, and static methods, a single 

195 shared cache is used. 

196 

197 All keyword arguments are passed through to ``functools.lru_cache``. 

198 

199 Cache management methods ``cache_info()`` and ``cache_clear()`` are 

200 available directly on the decorated function. For bound methods, 

201 these operate on the per-instance cache for the bound instance. 

202 """ 

203 

204 if func is None: 

205 return partial(lru_cache, **kwargs) 

206 

207 def _wrapper(wrapped, instance, args, _kwargs): 

208 return wrapped(*args, **_kwargs) 

209 

210 _wrapper._self_lru_kwargs = kwargs 

211 

212 return decorator(_wrapper, proxy=_LRUCacheFunctionWrapper)(func)