Coverage for src/jquantstats/_cache.py: 100%

1"""Slot-backed caching for frozen, slotted dataclasses. 

2 

3`Portfolio` is a frozen dataclass with ``slots=True``, so neither 

4`functools.cached_property` (needs ``__dict__``) nor plain attribute 

5assignment (frozen) works for memoising derived values. Instead, every cache 

6lives in an explicitly declared slot field that ``__post_init__`` initialises 

7to ``None``, and `cached_in_slot` fills it via ``object.__setattr__`` on 

8first access. 

9 

10Caching is not thread-safe: concurrent first accesses may compute the value 

11redundantly, but never produce incorrect results because every thread stores 

12the same deterministic value. 

13""" 

14 

15from __future__ import annotations 

16 

17import functools 

18from collections.abc import Callable 

19from typing import Any, TypeVar 

20 

21T = TypeVar("T") 

22 

23 

24def cached_in_slot(slot: str) -> Callable[[Callable[[Any], T]], Callable[[Any], T]]: 

25 """Cache a zero-argument method's result in the slot field named *slot*. 

26 

27 Apply below ``@property`` so the property getter is the wrapped function: 

28 

29 >>> class Prices: 

30 ... __slots__ = ("_calls", "_profits_cache") 

31 ... def __init__(self): 

32 ... self._calls = 0 

33 ... self._profits_cache = None # the slot must start as None 

34 ... @property 

35 ... @cached_in_slot("_profits_cache") 

36 ... def profits(self): 

37 ... self._calls += 1 

38 ... return [1, 2, 3] 

39 

40 The second access is served from the slot rather than recomputed: 

41 

42 >>> prices = Prices() 

43 >>> prices.profits, prices.profits 

44 ([1, 2, 3], [1, 2, 3]) 

45 >>> prices._calls 

46 1 

47 

48 Args: 

49 slot: Name of the declared slot field used as the cache. The field 

50 must be initialised to ``None`` before first access (Portfolio 

51 does this in ``__post_init__``); a ``None`` value means 

52 "not yet computed". 

53 

54 Returns: 

55 A decorator that wraps the getter with read-through caching. 

56 """ 

57 

58 def decorator(fn: Callable[[Any], T]) -> Callable[[Any], T]: 

59 """Wrap *fn* with read-through caching against the configured slot.""" 

60 

61 @functools.wraps(fn) 

62 def wrapper(self: Any) -> T: 

63 """Return the cached value, computing and storing it on first access.""" 

64 cache = getattr(self, slot, None) 

65 if cache is None: 

66 cache = fn(self) 

67 # Direct write is safe: the owner is a frozen, slotted 

68 # dataclass that declares every cache field, so 

69 # object.__setattr__ cannot fail here. 

70 object.__setattr__(self, slot, cache) 

71 return cache 

72 

73 return wrapper 

74 

75 return decorator