Coverage for jinja2_async_environment/caching/manager.py: 33%
136 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-09-03 14:09 -0700
1"""Cache manager for dependency injection and centralized cache control."""
3import typing as t
4from types import ModuleType
6from anyio import Path as AsyncPath
8from .strategies import AdaptiveCache, CacheWarmer, HierarchicalCache, LFUCache
9from .typed import TypedCache
11# Specialized type aliases for different cache types
12PackageSpecCache = TypedCache[tuple[t.Any, t.Any]]
13TemplateRootCache = TypedCache[AsyncPath | None]
14CompilationCache = TypedCache[str]
15ModuleCache = TypedCache[ModuleType]
18class CacheManager:
19 """Centralized cache management with dependency injection support.
21 This manager provides type-safe caches for different use cases while
22 allowing dependency injection and proper resource management.
23 """
25 def __init__(
26 self,
27 package_cache_size: int = 500,
28 template_cache_size: int = 1000,
29 compilation_cache_size: int = 2000,
30 module_cache_size: int = 200,
31 default_ttl: int = 300,
32 ):
33 """Initialize the cache manager.
35 Args:
36 package_cache_size: Maximum size for package spec cache
37 template_cache_size: Maximum size for template root cache
38 compilation_cache_size: Maximum size for compilation cache
39 module_cache_size: Maximum size for module import cache
40 default_ttl: Default TTL for all caches in seconds
41 """
42 # Initialize type-safe caches
43 self.package_cache: PackageSpecCache = TypedCache(
44 max_size=package_cache_size,
45 default_ttl=default_ttl * 6, # Package specs change rarely
46 )
48 self.template_cache: TemplateRootCache = TypedCache(
49 max_size=template_cache_size,
50 default_ttl=default_ttl * 6, # Template roots change rarely
51 )
53 self.compilation_cache: CompilationCache = TypedCache(
54 max_size=compilation_cache_size,
55 default_ttl=default_ttl * 2, # Compiled templates may change
56 )
58 self.module_cache: ModuleCache = TypedCache(
59 max_size=module_cache_size,
60 default_ttl=default_ttl * 12, # Modules change very rarely
61 )
63 self._default_ttl = default_ttl
65 def clear_all(self) -> None:
66 """Clear all caches."""
67 self.package_cache.clear()
68 self.template_cache.clear()
69 self.compilation_cache.clear()
70 self.module_cache.clear()
72 def cleanup_expired(self) -> dict[str, int]:
73 """Clean up expired entries from all caches.
75 Returns:
76 Dictionary with count of expired entries per cache
77 """
78 return {
79 "package_cache": self.package_cache.cleanup_expired(),
80 "template_cache": self.template_cache.cleanup_expired(),
81 "compilation_cache": self.compilation_cache.cleanup_expired(),
82 "module_cache": self.module_cache.cleanup_expired(),
83 }
85 def get_statistics(self) -> dict[str, dict[str, t.Any]]:
86 """Get statistics for all caches.
88 Returns:
89 Dictionary with statistics for each cache
90 """
91 return {
92 "package_cache": self.package_cache.get_statistics(),
93 "template_cache": self.template_cache.get_statistics(),
94 "compilation_cache": self.compilation_cache.get_statistics(),
95 "module_cache": self.module_cache.get_statistics(),
96 }
98 def resize_caches(
99 self,
100 package_size: int | None = None,
101 template_size: int | None = None,
102 compilation_size: int | None = None,
103 module_size: int | None = None,
104 ) -> None:
105 """Resize cache maximum sizes.
107 Args:
108 package_size: New size for package cache (None to keep current)
109 template_size: New size for template cache (None to keep current)
110 compilation_size: New size for compilation cache (None to keep current)
111 module_size: New size for module cache (None to keep current)
112 """
113 if package_size is not None:
114 self.package_cache.resize(package_size)
115 if template_size is not None:
116 self.template_cache.resize(template_size)
117 if compilation_size is not None:
118 self.compilation_cache.resize(compilation_size)
119 if module_size is not None:
120 self.module_cache.resize(module_size)
122 @classmethod
123 def get_default(cls) -> "CacheManager":
124 """Get the default global cache manager instance.
126 This provides backward compatibility while allowing dependency injection.
128 Returns:
129 Default cache manager instance
130 """
131 if not hasattr(cls, "_default_instance"):
132 cls._default_instance = cls()
133 return cls._default_instance
135 @classmethod
136 def set_default(cls, manager: "CacheManager") -> None:
137 """Set a new default cache manager instance.
139 Args:
140 manager: New cache manager to use as default
141 """
142 cls._default_instance = manager
144 def create_scoped_manager(
145 self,
146 package_ttl: int | None = None,
147 template_ttl: int | None = None,
148 compilation_ttl: int | None = None,
149 module_ttl: int | None = None,
150 ) -> "CacheManager":
151 """Create a new cache manager with different TTL settings.
153 Useful for creating isolated cache environments for testing
154 or different application contexts.
156 Args:
157 package_ttl: TTL for package cache (None to use default)
158 template_ttl: TTL for template cache (None to use default)
159 compilation_ttl: TTL for compilation cache (None to use default)
160 module_ttl: TTL for module cache (None to use default)
162 Returns:
163 New cache manager with specified TTL settings
164 """
165 # Use current cache sizes for the scoped manager
166 stats = self.get_statistics()
168 manager = CacheManager(
169 package_cache_size=stats["package_cache"]["max_size"],
170 template_cache_size=stats["template_cache"]["max_size"],
171 compilation_cache_size=stats["compilation_cache"]["max_size"],
172 module_cache_size=stats["module_cache"]["max_size"],
173 default_ttl=self._default_ttl,
174 )
176 # Override specific TTLs if provided
177 if package_ttl is not None:
178 manager.package_cache._default_ttl = package_ttl
179 if template_ttl is not None:
180 manager.template_cache._default_ttl = template_ttl
181 if compilation_ttl is not None:
182 manager.compilation_cache._default_ttl = compilation_ttl
183 if module_ttl is not None:
184 manager.module_cache._default_ttl = module_ttl
186 return manager
188 def get_memory_usage_estimate(self) -> dict[str, int]:
189 """Get rough memory usage estimates for all caches.
191 Returns:
192 Dictionary with estimated memory usage per cache in bytes
193 """
194 # Rough estimates based on typical object sizes
195 AVERAGE_SIZES = {
196 "package_spec": 200, # Loader + spec objects
197 "template_root": 100, # Path objects
198 "compilation": 5000, # Compiled template code
199 "module": 1000, # Module objects
200 }
202 stats = self.get_statistics()
204 memory_usage = {}
205 for cache_name in [
206 "package_cache",
207 "template_cache",
208 "compilation_cache",
209 "module_cache",
210 ]:
211 cache_stats = stats.get(cache_name, {})
212 if isinstance(cache_stats, dict) and "size" in cache_stats:
213 size = cache_stats["size"]
214 else:
215 # Handle hierarchical cache which might not have direct size
216 size = getattr(getattr(self, cache_name, None), "__len__", lambda: 0)()
217 if callable(size):
218 size = size()
220 avg_size = AVERAGE_SIZES.get(cache_name.replace("_cache", ""), 100)
221 memory_usage[cache_name] = size * avg_size
223 return memory_usage
225 def get(self, cache_type: str, key: str) -> t.Any:
226 """Get value from specified cache.
228 Args:
229 cache_type: Type of cache ('package', 'template', 'compilation', 'module')
230 key: Cache key
232 Returns:
233 Cached value or None if not found
235 Raises:
236 ValueError: If cache_type is not recognized
237 """
238 cache_map = {
239 "package": self.package_cache,
240 "template": self.template_cache,
241 "compilation": self.compilation_cache,
242 "module": self.module_cache,
243 }
245 if cache_type not in cache_map:
246 raise ValueError(f"Unknown cache type: {cache_type}")
248 return cache_map[cache_type].get(key)
250 def set(
251 self, cache_type: str, key: str, value: t.Any, ttl: int | None = None
252 ) -> None:
253 """Set value in specified cache.
255 Args:
256 cache_type: Type of cache ('package', 'template', 'compilation', 'module')
257 key: Cache key
258 value: Value to store
259 ttl: Time-to-live in seconds (uses cache default if None)
261 Raises:
262 ValueError: If cache_type is not recognized
263 """
264 cache_map = {
265 "package": self.package_cache,
266 "template": self.template_cache,
267 "compilation": self.compilation_cache,
268 "module": self.module_cache,
269 }
271 if cache_type not in cache_map:
272 raise ValueError(f"Unknown cache type: {cache_type}")
274 cache_map[cache_type].set(key, value, ttl)
276 def delete(self, cache_type: str, key: str) -> bool:
277 """Delete value from specified cache.
279 Args:
280 cache_type: Type of cache ('package', 'template', 'compilation', 'module')
281 key: Cache key
283 Returns:
284 True if key was deleted, False if not found
286 Raises:
287 ValueError: If cache_type is not recognized
288 """
289 cache_map = {
290 "package": self.package_cache,
291 "template": self.template_cache,
292 "compilation": self.compilation_cache,
293 "module": self.module_cache,
294 }
296 if cache_type not in cache_map:
297 raise ValueError(f"Unknown cache type: {cache_type}")
299 return cache_map[cache_type].delete(key)
301 def create_cache_warmer(self) -> CacheWarmer:
302 """Create a cache warmer for this cache manager.
304 Returns:
305 CacheWarmer instance for preloading caches
306 """
307 return CacheWarmer(self)
309 def __repr__(self) -> str:
310 """String representation of cache manager."""
311 stats = self.get_statistics()
312 total_size = sum(cache["size"] for cache in stats.values())
313 return f"CacheManager(total_entries={total_size}, caches=4)"
316class AdvancedCacheManager(CacheManager):
317 """Enhanced cache manager with advanced strategies and monitoring."""
319 def __init__(
320 self,
321 strategy: str = "adaptive", # "lru", "lfu", "adaptive", "hierarchical"
322 package_cache_size: int = 500,
323 template_cache_size: int = 1000,
324 compilation_cache_size: int = 2000,
325 module_cache_size: int = 200,
326 default_ttl: int = 300,
327 enable_hierarchical: bool = False,
328 l1_cache_size: int = 100,
329 ):
330 """Initialize advanced cache manager.
332 Args:
333 strategy: Cache strategy to use ("lru", "lfu", "adaptive", "hierarchical")
334 package_cache_size: Size for package cache
335 template_cache_size: Size for template cache
336 compilation_cache_size: Size for compilation cache
337 module_cache_size: Size for module cache
338 default_ttl: Default TTL in seconds
339 enable_hierarchical: Enable hierarchical caching for templates
340 l1_cache_size: L1 cache size for hierarchical mode
341 """
342 # Don't call super().__init__() - we'll create our own caches
343 self._strategy = strategy
344 self._enable_hierarchical = enable_hierarchical
345 self._default_ttl = default_ttl
347 # Create caches based on strategy
348 self._create_caches(
349 strategy,
350 package_cache_size,
351 template_cache_size,
352 compilation_cache_size,
353 module_cache_size,
354 default_ttl,
355 l1_cache_size,
356 )
358 def _create_caches(
359 self,
360 strategy: str,
361 package_size: int,
362 template_size: int,
363 compilation_size: int,
364 module_size: int,
365 ttl: int,
366 l1_size: int,
367 ) -> None:
368 """Create caches with specified strategy."""
369 cache_factory = {
370 "lru": lambda size, ttl: TypedCache(max_size=size, default_ttl=ttl),
371 "lfu": lambda size, ttl: LFUCache(max_size=size, default_ttl=ttl),
372 "adaptive": lambda size, ttl: AdaptiveCache(max_size=size, default_ttl=ttl),
373 }
375 factory = cache_factory.get(strategy, cache_factory["lru"])
377 # Create standard caches
378 self.package_cache = factory(package_size, ttl * 6)
379 self.compilation_cache = factory(compilation_size, ttl * 2)
380 self.module_cache = factory(module_size, ttl * 12)
382 # Create template cache (potentially hierarchical)
383 if self._enable_hierarchical:
384 self.template_cache = HierarchicalCache(
385 l1_size=l1_size, l2_size=template_size, l1_ttl=ttl, l2_ttl=ttl * 6
386 )
387 else:
388 self.template_cache = factory(template_size, ttl * 6)
390 def get_extended_statistics(self) -> dict[str, t.Any]:
391 """Get extended statistics for all caches."""
392 stats = self.get_statistics()
394 # Add strategy-specific statistics
395 extended_stats = {
396 "strategy": self._strategy,
397 "hierarchical_enabled": self._enable_hierarchical,
398 "base_stats": stats,
399 }
401 # Get advanced statistics from caches that support it
402 for cache_name, cache in [
403 ("package_cache", self.package_cache),
404 ("template_cache", self.template_cache),
405 ("compilation_cache", self.compilation_cache),
406 ("module_cache", self.module_cache),
407 ]:
408 if hasattr(cache, "get_extended_statistics"):
409 extended_stats[f"{cache_name}_extended"] = (
410 cache.get_extended_statistics()
411 )
412 elif hasattr(cache, "get_strategy_info"):
413 extended_stats[f"{cache_name}_strategy"] = cache.get_strategy_info()
415 return extended_stats
417 def optimize_caches(self) -> dict[str, t.Any]:
418 """Perform cache optimization and return results.
420 Returns:
421 Dictionary with optimization results
422 """
423 results = {}
425 # Cleanup expired entries
426 cleanup_results = self.cleanup_expired()
427 results["cleanup"] = cleanup_results
429 # Trigger strategy evaluation for adaptive caches
430 for cache_name, cache in [
431 ("package_cache", self.package_cache),
432 ("template_cache", self.template_cache),
433 ("compilation_cache", self.compilation_cache),
434 ("module_cache", self.module_cache),
435 ]:
436 if hasattr(cache, "_evaluate_strategy"):
437 old_strategy = getattr(cache, "_strategy", "unknown")
438 cache._evaluate_strategy()
439 new_strategy = getattr(cache, "_strategy", "unknown")
440 results[f"{cache_name}_strategy_change"] = {
441 "old": old_strategy,
442 "new": new_strategy,
443 "changed": old_strategy != new_strategy,
444 }
446 return results
448 def get_memory_efficiency_report(self) -> dict[str, t.Any]:
449 """Generate detailed memory efficiency report."""
450 stats = self.get_extended_statistics()
451 memory_usage = self.get_memory_usage_estimate()
453 total_memory = sum(memory_usage.values())
454 total_entries = sum(cache["size"] for cache in stats["base_stats"].values())
456 # Calculate efficiency metrics
457 efficiency_report = {
458 "total_memory_bytes": total_memory,
459 "total_entries": total_entries,
460 "avg_memory_per_entry": total_memory / total_entries
461 if total_entries > 0
462 else 0,
463 "memory_by_cache": memory_usage,
464 "cache_utilization": {},
465 "recommendations": [],
466 }
468 # Calculate utilization per cache
469 for cache_name, cache_stats in stats["base_stats"].items():
470 utilization = cache_stats["fill_ratio"]
471 efficiency_report["cache_utilization"][cache_name] = {
472 "fill_ratio": utilization,
473 "hit_rate": cache_stats["hit_rate"],
474 "efficiency_score": utilization * cache_stats["hit_rate"],
475 }
477 # Generate recommendations
478 if utilization < 0.3 and cache_stats["size"] > 0:
479 efficiency_report["recommendations"].append(
480 f"Consider reducing {cache_name} size (low utilization: {utilization:.1%})"
481 )
482 elif utilization > 0.9:
483 efficiency_report["recommendations"].append(
484 f"Consider increasing {cache_name} size (high utilization: {utilization:.1%})"
485 )
487 if cache_stats["hit_rate"] < 0.5 and cache_stats["size"] > 0:
488 efficiency_report["recommendations"].append(
489 f"Consider tuning {cache_name} TTL (low hit rate: {cache_stats['hit_rate']:.1%})"
490 )
492 return efficiency_report