Coverage for src / metaclass_registry / cache.py: 96%

1""" 

2Generic caching system for plugin registries. 

3 

4Provides unified caching for both function registries (Pattern B) and 

5metaclass registries (Pattern A), eliminating code duplication and 

6ensuring consistent cache behavior across the codebase. 

7 

8Architecture: 

9- RegistryCacheManager: Generic cache manager for any registry type 

10- Supports version validation, age-based invalidation, mtime checking 

11- JSON-based serialization with custom serializers/deserializers 

12- XDG-compliant cache locations 

13 

14Usage: 

15 # For function registries 

16 cache_mgr = RegistryCacheManager( 

17 cache_name="scikit_image_functions", 

18 version_getter=lambda: skimage.__version__, 

19 serializer=serialize_function_metadata, 

20 deserializer=deserialize_function_metadata 

21 ) 

22  

23 # For metaclass registries 

24 cache_mgr = RegistryCacheManager( 

25 cache_name="microscope_handlers", 

26 version_getter=lambda: openhcs.__version__, 

27 serializer=serialize_plugin_class, 

28 deserializer=deserialize_plugin_class 

29 ) 

30""" 

31 

32import json 

33import logging 

34import time 

35from pathlib import Path 

36from typing import Dict, Any, Optional, Callable, TypeVar, Generic 

37from dataclasses import dataclass 

38 

39logger = logging.getLogger(__name__) 

40 

41 

42def get_cache_file_path(cache_name: str) -> Path: 

43 """ 

44 Get XDG-compliant cache file path. 

45 

46 Args: 

47 cache_name: Name of the cache file 

48 

49 Returns: 

50 Path to cache file in XDG cache directory 

51 """ 

52 from . import _home 

53 

54 # Use XDG_CACHE_HOME if set, otherwise default to ~/.cache 

55 import os 

56 cache_home = os.environ.get('XDG_CACHE_HOME') 

57 if not cache_home: 

58 cache_home = Path(_home.get_home_dir()) / '.cache' 

59 else: 

60 cache_home = Path(cache_home) 

61 

62 # Create metaclass-registry subdirectory 

63 cache_dir = cache_home / 'metaclass-registry' 

64 cache_dir.mkdir(parents=True, exist_ok=True) 

65 

66 return cache_dir / cache_name 

67 

68T = TypeVar('T') # Generic type for cached items 

69 

70 

71@dataclass 

72class CacheConfig: 

73 """Configuration for registry caching behavior.""" 

74 max_age_days: int = 7 # Maximum cache age before invalidation 

75 check_mtimes: bool = False # Check file modification times 

76 cache_version: str = "1.0" # Cache format version 

77 

78 

79class RegistryCacheManager(Generic[T]): 

80 """ 

81 Generic cache manager for plugin registries. 

82  

83 Handles caching, validation, and reconstruction of registry data 

84 with support for version checking, age-based invalidation, and 

85 custom serialization. 

86  

87 Type Parameters: 

88 T: Type of items being cached (e.g., FunctionMetadata, Type[Plugin]) 

89 """ 

90 

91 def __init__( 

92 self, 

93 cache_name: str, 

94 version_getter: Callable[[], str], 

95 serializer: Callable[[T], Dict[str, Any]], 

96 deserializer: Callable[[Dict[str, Any]], T], 

97 config: Optional[CacheConfig] = None 

98 ): 

99 """ 

100 Initialize cache manager. 

101  

102 Args: 

103 cache_name: Name for the cache file (e.g., "microscope_handlers") 

104 version_getter: Function that returns current version string 

105 serializer: Function to serialize item to JSON-compatible dict 

106 deserializer: Function to deserialize dict back to item 

107 config: Optional cache configuration 

108 """ 

109 self.cache_name = cache_name 

110 self.version_getter = version_getter 

111 self.serializer = serializer 

112 self.deserializer = deserializer 

113 self.config = config or CacheConfig() 

114 self._cache_path = get_cache_file_path(f"{cache_name}.json") 

115 

116 def load_cache(self) -> Optional[Dict[str, T]]: 

117 """ 

118 Load cached items with validation. 

119  

120 Returns: 

121 Dictionary of cached items, or None if cache is invalid 

122 """ 

123 if not self._cache_path.exists(): 

124 logger.debug(f"No cache found for {self.cache_name}") 

125 return None 

126 

127 try: 

128 with open(self._cache_path, 'r') as f: 

129 cache_data = json.load(f) 

130 except json.JSONDecodeError: 

131 logger.warning(f"Corrupt cache file {self._cache_path}, rebuilding") 

132 self._cache_path.unlink(missing_ok=True) 

133 return None 

134 

135 # Validate cache version 

136 if cache_data.get('cache_version') != self.config.cache_version: 

137 logger.debug(f"Cache version mismatch for {self.cache_name}") 

138 return None 

139 

140 # Validate library/package version 

141 cached_version = cache_data.get('version', 'unknown') 

142 current_version = self.version_getter() 

143 if cached_version != current_version: 

144 logger.info( 

145 f"{self.cache_name} version changed " 

146 f"({cached_version} → {current_version}) - cache invalid" 

147 ) 

148 return None 

149 

150 # Validate cache age 

151 cache_timestamp = cache_data.get('timestamp', 0) 

152 cache_age_days = (time.time() - cache_timestamp) / (24 * 3600) 

153 if cache_age_days > self.config.max_age_days: 

154 logger.debug( 

155 f"Cache for {self.cache_name} is {cache_age_days:.1f} days old - rebuilding" 

156 ) 

157 return None 

158 

159 # Validate file mtimes if configured 

160 if self.config.check_mtimes and 'file_mtimes' in cache_data: 

161 if not self._validate_mtimes(cache_data['file_mtimes']): 

162 logger.debug(f"File modifications detected for {self.cache_name}") 

163 return None 

164 

165 # Deserialize items 

166 items = {} 

167 for key, item_data in cache_data.get('items', {}).items(): 

168 try: 

169 items[key] = self.deserializer(item_data) 

170 except Exception as e: 

171 logger.warning(f"Failed to deserialize {key} from cache: {e}") 

172 return None # Invalidate entire cache on any deserialization error 

173 

174 logger.info(f"✅ Loaded {len(items)} items from {self.cache_name} cache") 

175 return items 

176 

177 def save_cache( 

178 self, 

179 items: Dict[str, T], 

180 file_mtimes: Optional[Dict[str, float]] = None 

181 ) -> None: 

182 """ 

183 Save items to cache. 

184  

185 Args: 

186 items: Dictionary of items to cache 

187 file_mtimes: Optional dict of file paths to modification times 

188 """ 

189 cache_data = { 

190 'cache_version': self.config.cache_version, 

191 'version': self.version_getter(), 

192 'timestamp': time.time(), 

193 'items': {} 

194 } 

195 

196 # Add file mtimes if provided 

197 if file_mtimes: 

198 cache_data['file_mtimes'] = file_mtimes 

199 

200 # Serialize items 

201 for key, item in items.items(): 

202 try: 

203 cache_data['items'][key] = self.serializer(item) 

204 except Exception as e: 

205 logger.warning(f"Failed to serialize {key} for cache: {e}") 

206 

207 # Save to disk 

208 try: 

209 self._cache_path.parent.mkdir(parents=True, exist_ok=True) 

210 with open(self._cache_path, 'w') as f: 

211 json.dump(cache_data, f, indent=2) 

212 logger.info(f"💾 Saved {len(items)} items to {self.cache_name} cache") 

213 except Exception as e: 

214 logger.warning(f"Failed to save {self.cache_name} cache: {e}") 

215 

216 def clear_cache(self) -> None: 

217 """Clear the cache file.""" 

218 if self._cache_path.exists(): 

219 self._cache_path.unlink() 

220 logger.info(f"🧹 Cleared {self.cache_name} cache") 

221 

222 def _validate_mtimes(self, cached_mtimes: Dict[str, float]) -> bool: 

223 """ 

224 Validate that file modification times haven't changed. 

225  

226 Args: 

227 cached_mtimes: Dictionary of file paths to cached mtimes 

228  

229 Returns: 

230 True if all mtimes match, False if any file changed 

231 """ 

232 for file_path, cached_mtime in cached_mtimes.items(): 

233 path = Path(file_path) 

234 if not path.exists(): 

235 return False # File was deleted 

236 

237 current_mtime = path.stat().st_mtime 

238 if abs(current_mtime - cached_mtime) > 1.0: # 1 second tolerance 

239 return False # File was modified 

240 

241 return True 

242 

243 

244# Serializers for metaclass registries (Pattern A) 

245 

246def serialize_plugin_class(plugin_class: type) -> Dict[str, Any]: 

247 """ 

248 Serialize a plugin class to JSON-compatible dict. 

249  

250 Args: 

251 plugin_class: Plugin class to serialize 

252  

253 Returns: 

254 Dictionary with module and class name 

255 """ 

256 return { 

257 'module': plugin_class.__module__, 

258 'class_name': plugin_class.__name__, 

259 'qualname': plugin_class.__qualname__ 

260 } 

261 

262 

263def deserialize_plugin_class(data: Dict[str, Any]) -> type: 

264 """ 

265 Deserialize a plugin class from JSON-compatible dict. 

266  

267 Args: 

268 data: Dictionary with module and class name 

269  

270 Returns: 

271 Reconstructed plugin class 

272  

273 Raises: 

274 ImportError: If module cannot be imported 

275 AttributeError: If class not found in module 

276 """ 

277 import importlib 

278 

279 module = importlib.import_module(data['module']) 

280 plugin_class = getattr(module, data['class_name']) 

281 return plugin_class 

282 

283 

284def get_package_file_mtimes(package_path: str) -> Dict[str, float]: 

285 """ 

286 Get modification times for all Python files in a package. 

287  

288 Args: 

289 package_path: Package path (e.g., "openhcs.microscopes") 

290  

291 Returns: 

292 Dictionary mapping file paths to modification times 

293 """ 

294 import importlib 

295 from pathlib import Path 

296 

297 try: 

298 pkg = importlib.import_module(package_path) 

299 pkg_dir = Path(pkg.__file__).parent 

300 

301 mtimes = {} 

302 for py_file in pkg_dir.rglob("*.py"): 

303 if not py_file.name.startswith('_'): # Skip __pycache__, etc. 

304 mtimes[str(py_file)] = py_file.stat().st_mtime 

305 

306 return mtimes 

307 except Exception as e: 

308 logger.warning(f"Failed to get mtimes for {package_path}: {e}") 

309 return {} 

310