Coverage for jinja2_async_environment/caching/unified.py: 29%

1"""Unified cache implementation for backward compatibility. 

2 

3This module provides the UnifiedCache class that maintains compatibility 

4with the existing cache interface while leveraging the new type-safe 

5caching infrastructure. 

6""" 

7 

8import typing as t 

9from threading import RLock 

10 

11from .typed import TypedCache 

12 

13 

14class UnifiedCache: 

15 """Unified cache system for backward compatibility. 

16 

17 This cache maintains the existing API while internally using 

18 the new type-safe cache infrastructure. It will be gradually 

19 phased out in favor of the CacheManager approach. 

20 """ 

21 

22 def __init__(self, default_ttl: int = 300) -> None: 

23 """Initialize the unified cache. 

24 

25 Args: 

26 default_ttl: Default time-to-live in seconds 

27 """ 

28 # Internal type-safe caches - using Any for backward compatibility 

29 self._caches: dict[str, TypedCache[t.Any]] = { 

30 "package_import": TypedCache(max_size=200, default_ttl=default_ttl * 12), 

31 "package_spec": TypedCache(max_size=500, default_ttl=default_ttl * 6), 

32 "template_root": TypedCache(max_size=1000, default_ttl=default_ttl * 6), 

33 } 

34 

35 self._default_ttl = default_ttl 

36 self._lock = RLock() 

37 

38 def get(self, cache_type: str, key: t.Any, default: t.Any = None) -> t.Any: 

39 """Get a value from the specified cache. 

40 

41 Args: 

42 cache_type: Type of cache ("package_import", "package_spec", "template_root") 

43 key: Cache key 

44 default: Default value if not found 

45 

46 Returns: 

47 Cached value or default 

48 """ 

49 with self._lock: 

50 if cache_type not in self._caches: 

51 return default 

52 

53 # Convert key to string for internal cache 

54 str_key = str(key) if not isinstance(key, str) else key 

55 

56 value = self._caches[cache_type].get(str_key) 

57 return value if value is not None else default 

58 

59 def set( 

60 self, cache_type: str, key: t.Any, value: t.Any, ttl: int | None = None 

61 ) -> None: 

62 """Store a value in the specified cache. 

63 

64 Args: 

65 cache_type: Type of cache 

66 key: Cache key 

67 value: Value to store 

68 ttl: Time-to-live in seconds (uses default if None) 

69 """ 

70 with self._lock: 

71 if cache_type not in self._caches: 

72 # Dynamically create cache for unknown types 

73 self._caches[cache_type] = TypedCache( 

74 max_size=1000, default_ttl=ttl or self._default_ttl 

75 ) 

76 

77 # Convert key to string for internal cache 

78 str_key = str(key) if not isinstance(key, str) else key 

79 

80 self._caches[cache_type].set(str_key, value, ttl) 

81 

82 def clear(self, cache_type: str | None = None) -> None: 

83 """Clear cache entries. 

84 

85 Args: 

86 cache_type: Specific cache to clear (None for all caches) 

87 """ 

88 with self._lock: 

89 if cache_type is None: 

90 # Clear all caches 

91 for cache in self._caches.values(): 

92 cache.clear() 

93 elif cache_type in self._caches: 

94 self._caches[cache_type].clear() 

95 

96 def clear_all(self) -> None: 

97 """Clear all cache entries.""" 

98 self.clear() 

99 

100 def cleanup_expired(self) -> None: 

101 """Remove expired entries from all caches.""" 

102 with self._lock: 

103 for cache in self._caches.values(): 

104 cache.cleanup_expired() 

105 

106 def get_statistics(self) -> dict[str, dict[str, t.Any]]: 

107 """Get statistics for all caches. 

108 

109 Returns: 

110 Dictionary with statistics for each cache type 

111 """ 

112 with self._lock: 

113 stats = {} 

114 for cache_type, cache in self._caches.items(): 

115 stats[cache_type] = cache.get_statistics() 

116 return stats 

117 

118 def _is_valid(self, cache_type: str, key: t.Any) -> bool: 

119 """Check if a cache entry is valid (exists and not expired). 

120 

121 Args: 

122 cache_type: Type of cache 

123 key: Cache key 

124 

125 Returns: 

126 True if entry is valid, False otherwise 

127 """ 

128 with self._lock: 

129 if cache_type not in self._caches: 

130 return False 

131 

132 str_key = str(key) if not isinstance(key, str) else key 

133 return self._caches[cache_type].contains(str_key) 

134 

135 def resize_cache(self, cache_type: str, new_size: int) -> None: 

136 """Resize a specific cache. 

137 

138 Args: 

139 cache_type: Type of cache to resize 

140 new_size: New maximum size 

141 """ 

142 with self._lock: 

143 if cache_type in self._caches: 

144 self._caches[cache_type].resize(new_size) 

145 

146 def get_cache_types(self) -> list[str]: 

147 """Get list of available cache types. 

148 

149 Returns: 

150 List of cache type names 

151 """ 

152 with self._lock: 

153 return list(self._caches.keys()) 

154 

155 def contains(self, cache_type: str, key: t.Any) -> bool: 

156 """Check if a key exists in the specified cache. 

157 

158 Args: 

159 cache_type: Type of cache 

160 key: Cache key 

161 

162 Returns: 

163 True if key exists and is valid, False otherwise 

164 """ 

165 return self._is_valid(cache_type, key) 

166 

167 def __len__(self) -> int: 

168 """Get total number of entries across all caches.""" 

169 with self._lock: 

170 return sum(len(cache) for cache in self._caches.values()) 

171 

172 def __repr__(self) -> str: 

173 """String representation of unified cache.""" 

174 with self._lock: 

175 total_entries = len(self) 

176 return f"UnifiedCache(types={len(self._caches)}, entries={total_entries})" 

177 

178 

179# Backward compatibility function 

180def _clear_expired_cache() -> None: 

181 """Clear expired cache entries (backward compatibility function).""" 

182 # This will be connected to the global unified cache instance 

183 # when it's available through the loader context 

184 pass