Coverage for flogin/caching.py: 100%

1# ruff: noqa: ANN202 

2 

3from __future__ import annotations 

4 

5from collections import defaultdict 

6from collections.abc import ( 

7 AsyncGenerator, 

8 AsyncIterator, 

9 Awaitable, 

10 Callable, 

11 Coroutine, 

12 Hashable, 

13) 

14from functools import _make_key as make_cached_key 

15from typing import ( 

16 TYPE_CHECKING, 

17 Any, 

18 Generic, 

19 ParamSpec, 

20 Self, 

21 TypeVar, 

22 overload, 

23) 

24 

25from .utils import MISSING, decorator 

26 

27Coro = TypeVar("Coro", bound=Callable[..., Coroutine[Any, Any, Any]]) 

28AGenT = TypeVar("AGenT", bound=Callable[..., AsyncGenerator[Any, Any]]) 

29 

30if TYPE_CHECKING: 

31 from typing_extensions import ParamSpec, TypeVar # noqa: TC004 

32 

33 T = TypeVar("T", default=Any) 

34 RT = TypeVar("RT", default=Any) 

35 CT = TypeVar("CT", default=Any) 

36 P = ParamSpec("P", default=...) 

37else: 

38 T = TypeVar("T") 

39 RT = TypeVar("RT") 

40 CT = TypeVar("CT") 

41 P = ParamSpec("P") 

42 

43 

44__all__ = ( 

45 "cached_callable", 

46 "cached_coro", 

47 "cached_gen", 

48 "cached_property", 

49 "clear_cache", 

50) 

51 

52__cached_objects__: defaultdict[Any, list[BaseCachedObject]] = defaultdict(list) 

53 

54 

55def clear_cache(key: str | None = MISSING) -> None: 

56 r"""This function is used to clear the cache of items that have been cached with this module. 

57 

58 The caching decorators provide an optional positional argument that acts as a ``name`` argument, which is used in combination of this function. 

59 

60 Parameters 

61 ---------- 

62 key: Optional[:class:`str` | ``None``] 

63 If :class:`str` is passed, every cached item with a name equal to ``key`` will have their cache cleared. If ``None`` is passed, every cached item with a name equal to ``None`` will have their cache cleared (default value for a cached item's name is ``None``). Lastly, if the ``key`` parameter is not passed at all, all caches will be cleared. 

64 """ 

65 

66 if key is MISSING: 

67 items: list[BaseCachedObject] = [] 

68 for section in __cached_objects__.values(): 

69 items.extend(section) 

70 else: 

71 items = __cached_objects__[key] 

72 

73 for cached_obj in items: 

74 cached_obj.clear_cache() 

75 

76 

77class BaseCachedObject(Generic[RT, CT, P]): 

78 def __init__(self, obj: Callable[P, RT], name: str | None = None) -> None: 

79 self.obj = obj 

80 self.name = name or obj.__name__ 

81 self.cache: dict[Hashable, CT] = {} 

82 __cached_objects__[name].append(self) 

83 

84 def __call__(self, *args: P.args, **kwargs: P.kwargs) -> RT: 

85 key = make_cached_key(args, kwargs, False) 

86 return self.call(key, *args, **kwargs) 

87 

88 def call(self, key: Hashable, *args: P.args, **kwargs: P.kwargs) -> RT: 

89 raise NotImplementedError 

90 

91 def clear_cache(self) -> None: 

92 self.cache.clear() 

93 

94 

95class CachedCoro(BaseCachedObject[Awaitable[T], T, P], Generic[T, P]): 

96 async def call(self, key: Hashable, *args: P.args, **kwargs: P.kwargs): 

97 try: 

98 return self.cache[key] 

99 except KeyError: 

100 self.cache[key] = await self.obj(*args, **kwargs) 

101 return self.cache[key] 

102 

103 

104class CachedGen(BaseCachedObject[AsyncIterator[T], list[T], P], Generic[T, P]): 

105 async def call(self, key: Hashable, *args: P.args, **kwargs: P.kwargs): 

106 try: 

107 for item in self.cache[key]: 

108 yield item 

109 except KeyError: 

110 self.cache[key] = [val async for val in self.obj(*args, **kwargs)] 

111 for item in self.cache[key]: 

112 yield item 

113 

114 

115class CachedProperty(BaseCachedObject, Generic[T]): 

116 value: T 

117 

118 @overload 

119 def __get__(self, instance: None, owner: type[Any] | None = None) -> Self: ... 

120 

121 @overload 

122 def __get__(self, instance: object, owner: type[Any] | None = None) -> T: ... 

123 

124 def __get__( 

125 self, instance: object | None, owner: type[Any] | None = None 

126 ) -> T | Self: 

127 if instance is None: 

128 return self 

129 

130 try: 

131 return self.value 

132 except AttributeError: 

133 self.value = self.obj(instance) 

134 return self.value 

135 

136 def clear_cache(self): 

137 try: 

138 del self.value 

139 except AttributeError: 

140 pass 

141 

142 

143class CachedCallable(BaseCachedObject[T, T, P], Generic[T, P]): 

144 def call(self, key: Hashable, *args: P.args, **kwargs: P.kwargs): 

145 try: 

146 return self.cache[key] 

147 except KeyError: 

148 self.cache[key] = self.obj(*args, **kwargs) 

149 return self.cache[key] 

150 

151 

152def _cached_deco(cls: type[BaseCachedObject], doc: str | None = None): 

153 def inner(obj: str | Callable[..., Any] | None = None): 

154 if isinstance(obj, str) or obj is None: 

155 

156 def inner(obj2: Callable[..., Any]): 

157 return cls(obj2, obj) 

158 

159 return inner 

160 return cls(obj) 

161 

162 inner.__doc__ = doc 

163 return decorator(inner) 

164 

165 

166CoroT = TypeVar("CoroT", bound=Callable[..., Awaitable[Any]]) 

167GenT = TypeVar("GenT", bound=Callable[..., AsyncGenerator[Any, Any]]) 

168CallableT = TypeVar("CallableT", bound=Callable[..., Any]) 

169 

170 

171@overload 

172def cached_coro(obj: str | None = None) -> Callable[[CoroT], CoroT]: ... 

173 

174 

175@overload 

176def cached_coro(obj: CoroT) -> CoroT: ... 

177 

178 

179@decorator 

180def cached_coro(obj: str | None | CoroT = None) -> Callable[[CoroT], CoroT] | CoroT: 

181 r"""A decorator to cache a coroutine's contents based on the passed arguments. This decorator can also be called with the optional positional argument acting as a ``name`` argument. This is useful when using :func:`~flogin.caching.clear_cache` as it lets you choose which items you want to clear the cache of. 

182 

183 .. NOTE:: 

184 The arguments passed to the coroutine must be hashable. 

185 

186 Example 

187 -------- 

188 .. code-block:: python3 

189 

190 @plugin.search() 

191 @cached_coro 

192 async def handler(query): 

193 ... 

194 

195 .. code-block:: python3 

196 

197 @plugin.search() 

198 @cached_coro("search-handler") 

199 async def handler(query): 

200 ... 

201 """ 

202 ... # cov: skip 

203 

204 

205@overload 

206def cached_gen(obj: str | None = None) -> Callable[[GenT], GenT]: ... 

207 

208 

209@overload 

210def cached_gen(obj: GenT) -> GenT: ... 

211 

212 

213@decorator 

214def cached_gen(obj: str | GenT | None = None) -> Callable[[GenT], GenT] | GenT: 

215 r"""A decorator to cache the contents of an async generator based on the passed arguments. This decorator can also be called with the optional positional argument acting as a ``name`` argument. This is useful when using :func:`~flogin.caching.clear_cache` as it lets you choose which items you want to clear the cache of. 

216 

217 .. NOTE:: 

218 The arguments passed to the generator must be hashable. 

219 

220 Example 

221 -------- 

222 .. code-block:: python3 

223 

224 @plugin.search() 

225 .cached_gen 

226 async def handler(query): 

227 ... 

228 

229 .. code-block:: python3 

230 

231 @plugin.search() 

232 @cached_gen("search-handler") 

233 async def handler(query): 

234 ... 

235 """ 

236 ... # cov: skip 

237 

238 

239@overload 

240def cached_property( 

241 obj: str | None = None, 

242) -> Callable[[Callable[[Any], T]], CachedProperty[T]]: ... 

243 

244 

245@overload 

246def cached_property(obj: Callable[[Any], T]) -> CachedProperty[T]: ... 

247 

248 

249@decorator 

250def cached_property( 

251 obj: str | Callable[[Any], T] | None = None, 

252) -> Callable[[Callable[[Any], T]], CachedProperty[T]] | CachedProperty[T]: 

253 r"""A decorator that is similar to the builtin `functools.cached_property <https://docs.python.org/3/library/functools.html#functools.cached_property>`__ decorator, but is async-safe and implements the ability to use :func:`~flogin.caching.clear_cache`. 

254 

255 This decorator can also be called with the optional positional argument acting as a ``name`` argument. This is useful when using :func:`~flogin.caching.clear_cache` as it lets you choose which items you want to clear the cache of. 

256 

257 Example 

258 -------- 

259 .. code-block:: python3 

260 

261 class X: 

262 @cached_property 

263 def test(self): 

264 ... 

265 

266 .. code-block:: python3 

267 

268 class X: 

269 @cached_property("test_prop") 

270 def test(self): 

271 ... 

272 """ 

273 

274 return _cached_deco(CachedProperty)(obj) # type: ignore[reportReturnType] 

275 

276 

277@overload 

278def cached_callable(obj: str | None = None) -> Callable[[CallableT], CallableT]: ... 

279 

280 

281@overload 

282def cached_callable(obj: CallableT) -> CallableT: ... 

283 

284 

285@decorator 

286def cached_callable( 

287 obj: str | CallableT | None = None, 

288) -> CallableT | Callable[[CallableT], CallableT]: 

289 r"""A decorator to cache a callable's output based on the passed arguments. This decorator can also be called with the optional positional argument acting as a ``name`` argument. This is useful when using :func:`~flogin.caching.clear_cache` as it lets you choose which items you want to clear the cache of. 

290 

291 .. NOTE:: 

292 The arguments passed to the callable must be hashable. 

293 

294 Example 

295 -------- 

296 .. code-block:: python3 

297 

298 @cached_callable 

299 def foo(bar): 

300 ... 

301 

302 .. code-block:: python3 

303 

304 @cached_callable("search-handler") 

305 def foo(bar): 

306 ... 

307 """ 

308 ... # cov: skip 

309 

310 

311if not TYPE_CHECKING: 

312 cached_coro = _cached_deco(CachedCoro, cached_coro.__doc__) 

313 cached_gen = _cached_deco(CachedGen, cached_gen.__doc__) 

314 cached_callable = _cached_deco(CachedCallable, cached_callable.__doc__)