Coverage for src/objectstate/token_cache.py: 32%

1""" 

2Token-based cache invalidation service. 

3 

4Provides a reusable abstraction for caching values that should be invalidated 

5when a global token changes (e.g., when any form value changes). 

6""" 

7 

8from typing import TypeVar, Generic, Optional, Callable, Tuple, Any, Dict 

9from dataclasses import dataclass 

10 

11T = TypeVar('T') 

12 

13 

14@dataclass(frozen=True) 

15class CacheKey: 

16 """Immutable cache key that can include multiple components.""" 

17 components: Tuple[Any, ...] 

18 

19 def __hash__(self): 

20 return hash(self.components) 

21 

22 def __eq__(self, other): 

23 if not isinstance(other, CacheKey): 

24 return False 

25 return self.components == other.components 

26 

27 @classmethod 

28 def from_args(cls, *args) -> 'CacheKey': 

29 """Create cache key from variable arguments.""" 

30 return cls(components=args) 

31 

32 

33class TokenCache(Generic[T]): 

34 """ 

35 Generic token-based cache with automatic invalidation. 

36  

37 The cache is invalidated when the token changes. This is useful for caching 

38 values that depend on global state (e.g., form values) that can change. 

39  

40 Example: 

41 # Create cache with token provider 

42 cache = TokenCache(lambda: ParameterFormManager._live_context_token_counter) 

43  

44 # Get or compute value 

45 value = cache.get_or_compute( 

46 key=CacheKey.from_args('scope', 'param_name'), 

47 compute_fn=lambda: expensive_computation() 

48 ) 

49  

50 # Cache is automatically invalidated when token changes 

51 """ 

52 

53 def __init__(self, token_provider: Callable[[], int]): 

54 """ 

55 Initialize token cache. 

56  

57 Args: 

58 token_provider: Function that returns the current token value 

59 """ 

60 self._token_provider = token_provider 

61 self._cache: Dict[CacheKey, T] = {} 

62 self._last_token: int = -1 

63 

64 def get_or_compute(self, key: CacheKey, compute_fn: Callable[[], T]) -> T: 

65 """ 

66 Get cached value or compute and cache it. 

67  

68 Args: 

69 key: Cache key 

70 compute_fn: Function to compute value if cache miss 

71  

72 Returns: 

73 Cached or computed value 

74 """ 

75 current_token = self._token_provider() 

76 

77 # Invalidate entire cache if token changed 

78 if current_token != self._last_token: 

79 self._cache.clear() 

80 self._last_token = current_token 

81 

82 # Check cache 

83 if key in self._cache: 

84 return self._cache[key] 

85 

86 # Compute and cache 

87 value = compute_fn() 

88 self._cache[key] = value 

89 return value 

90 

91 def invalidate(self): 

92 """Manually invalidate the entire cache.""" 

93 self._cache.clear() 

94 self._last_token = -1 

95 

96 def get(self, key: CacheKey) -> Optional[T]: 

97 """ 

98 Get cached value without computing. 

99  

100 Args: 

101 key: Cache key 

102  

103 Returns: 

104 Cached value or None if not found or token changed 

105 """ 

106 current_token = self._token_provider() 

107 

108 # Invalidate if token changed 

109 if current_token != self._last_token: 

110 self._cache.clear() 

111 self._last_token = current_token 

112 return None 

113 

114 return self._cache.get(key) 

115 

116 def put(self, key: CacheKey, value: T): 

117 """ 

118 Put value in cache. 

119  

120 Args: 

121 key: Cache key 

122 value: Value to cache 

123 """ 

124 current_token = self._token_provider() 

125 

126 # Update token if changed 

127 if current_token != self._last_token: 

128 self._cache.clear() 

129 self._last_token = current_token 

130 

131 self._cache[key] = value 

132 

133 

134class SingleValueTokenCache(Generic[T]): 

135 """ 

136 Simplified token cache for single values (no key needed). 

137  

138 Useful when you only need to cache one value that depends on a token. 

139  

140 Example: 

141 cache = SingleValueTokenCache(lambda: token_counter) 

142 value = cache.get_or_compute(lambda: expensive_computation()) 

143 """ 

144 

145 def __init__(self, token_provider: Callable[[], int]): 

146 """ 

147 Initialize single-value token cache. 

148  

149 Args: 

150 token_provider: Function that returns the current token value 

151 """ 

152 self._token_provider = token_provider 

153 self._cached_value: Optional[T] = None 

154 self._cached_token: int = -1 

155 

156 def get_or_compute(self, compute_fn: Callable[[], T]) -> T: 

157 """ 

158 Get cached value or compute and cache it. 

159  

160 Args: 

161 compute_fn: Function to compute value if cache miss 

162  

163 Returns: 

164 Cached or computed value 

165 """ 

166 current_token = self._token_provider() 

167 

168 # Check cache 

169 if current_token == self._cached_token and self._cached_value is not None: 

170 return self._cached_value 

171 

172 # Compute and cache 

173 value = compute_fn() 

174 self._cached_value = value 

175 self._cached_token = current_token 

176 return value 

177 

178 def invalidate(self): 

179 """Manually invalidate the cache.""" 

180 self._cached_value = None 

181 self._cached_token = -1 

182