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