Module

utils.cache_registry

Centralized cache registry for cache lifecycle management.

Provides a registry pattern where caches can register themselves with metadata for automatic cleanup, dependency tracking, and coordinated invalidation. This prevents memory leaks in tests and ensures correct cache invalidation order during builds.

Key Features:

  • Reason-based invalidation: Caches declare which events should trigger them
  • Dependency tracking: Caches can depend on other caches (cascade invalidation)
  • Topological sorting: Ensures correct invalidation order
  • Invalidation logging: Track what was invalidated and why

Usage:

from bengal.utils.cache_registry import (
    register_cache,
    invalidate_for_reason,
    InvalidationReason,
)

# Register a cache with lifecycle metadata
register_cache(
    "nav_tree",
    NavTreeCache.invalidate,
    invalidate_on={InvalidationReason.CONFIG_CHANGED, InvalidationReason.STRUCTURAL_CHANGE},
    depends_on={"build_cache"},
)

# Invalidate all caches that should respond to a config change
invalidated = invalidate_for_reason(InvalidationReason.CONFIG_CHANGED)
logger.debug("caches_invalidated", caches=invalidated)

For caches keyed by object IDs (like Site objects), use WeakKeyDictionary instead of regular dict to prevent memory leaks:

from weakref import WeakKeyDictionary

_my_cache: WeakKeyDictionary[Site, Any] = WeakKeyDictionary()
# No registration needed - automatically cleaned up when objects are GC'd

Related Modules:

  • bengal.tests.conftest: Uses clear_all_caches() in test fixtures
  • bengal.rendering.context: Registers global context cache
  • bengal.rendering.pipeline.thread_local: Registers parser cache
  • bengal.core.nav_tree: Registers NavTreeCache
2Classes12Functions

Classes

InvalidationReason 0 ▼
Why a cache was invalidated. Used to declare which events should trigger cache invalidation. Cache…

Why a cache was invalidated.

Used to declare which events should trigger cache invalidation. Caches register which reasons they respond to, enabling coordinated invalidation without scattered manual calls.

CacheEntry 4 ▼
Registered cache with metadata. Stores cache clear function along with lifecycle metadata for coor…

Registered cache with metadata.

Stores cache clear function along with lifecycle metadata for coordinated invalidation and dependency tracking.

Attributes

Name Type Description
name str

Unique cache name (for debugging and dependency tracking)
clear_fn Callable[[], None]

Callable that clears the cache (e.g., lambda: cache.clear())
invalidate_on set[InvalidationReason]

Set of reasons that should trigger invalidation
depends_on set[str]

Set of cache names this cache depends on

Functions

register_cache 4 None ▼
Register a cache with lifecycle metadata. Caches should register themselves at…
def register_cache(name: str, clear_fn: Callable[[], None], invalidate_on: set[InvalidationReason] | None = None, depends_on: set[str] | None = None) -> None

Register a cache with lifecycle metadata.

Caches should register themselves at module import time (standard pattern). This ensures all caches are registered before any build operations.

Parameters
Name Type Description
name str

Unique cache name (for debugging and dependency tracking)
clear_fn Callable[[], None]

Callable that clears the cache (e.g., lambda: cache.clear())
invalidate_on set[InvalidationReason] | None

Set of reasons that should trigger invalidation. Defaults to {FULL_REBUILD} if not specified.

Default:None
depends_on set[str] | None

Set of cache names this cache depends on (for cascade invalidation). Dependencies must already be registered.

Default:None
_validate_no_cycles 2 None ▼
Validate that adding new_cache with new_deps doesn't create a cycle. Uses DFS …
def _validate_no_cycles(new_cache: str, new_deps: set[str]) -> None

Validate that adding new_cache with new_deps doesn't create a cycle.

Uses DFS to detect cycles in dependency graph.

Parameters
Name Type Description
new_cache str

Name of cache being registered
new_deps set[str]

Dependencies of the new cache
_topological_sort 1 list[str] ▼
Topologically sort cache names by dependency order. Uses graphlib.TopologicalS…
def _topological_sort(cache_names: set[str]) -> list[str]

Topologically sort cache names by dependency order.

Uses graphlib.TopologicalSorter (Python 3.9+) for reliable ordering. Ensures dependencies are invalidated before dependents.

Parameters
Name Type Description
cache_names set[str]

Set of cache names to sort
Returns
list[str]
invalidate_for_reason 1 list[str] ▼
Invalidate all caches that should be cleared for this reason. This is the prim…
def invalidate_for_reason(reason: InvalidationReason) -> list[str]

Invalidate all caches that should be cleared for this reason.

This is the primary API for triggering cache invalidation. Instead of calling individual cache invalidation functions, call this with the appropriate reason and let caches self-select.

Parameters
Name Type Description
reason InvalidationReason

Why invalidation is occurring
Returns
list[str]
invalidate_with_dependents 2 list[str] ▼
Invalidate a specific cache and all caches that depend on it. Uses topological…
def invalidate_with_dependents(cache_name: str, reason: InvalidationReason) -> list[str]

Invalidate a specific cache and all caches that depend on it.

Uses topological sort to ensure correct order (dependencies before dependents).

Parameters
Name Type Description
cache_name str

Name of cache to invalidate
reason InvalidationReason

Reason for invalidation (for logging)
Returns
list[str]
_log_invalidation 3 None ▼
Log an invalidation event (internal helper).
def _log_invalidation(name: str, reason: InvalidationReason, timestamp: float) -> None
Parameters
Name Type Description
name str
reason InvalidationReason
timestamp float
unregister_cache 1 None ▼
Unregister a cache (rarely needed).
def unregister_cache(name: str) -> None
Parameters
Name Type Description
name str

Name of cache to unregister
clear_all_caches 0 None ▼
Clear all registered caches. This is the main function to call in test fixture…
def clear_all_caches() -> None

Clear all registered caches.

This is the main function to call in test fixtures or between builds to prevent memory leaks. Equivalent to invalidate_for_reason(FULL_REBUILD) but clears ALL caches regardless of their invalidate_on settings.

Thread Safety:

Thread-safe - clears all caches under lock.
list_registered_caches 0 list[str] ▼
List all registered cache names (for debugging).
def list_registered_caches() -> list[str]
Returns
list[str]
get_cache_info 1 dict | None ▼
Get information about a registered cache.
def get_cache_info(name: str) -> dict | None
Parameters
Name Type Description
name str

Cache name
Returns
dict | None
get_invalidation_log 0 list[tuple[str, str, flo… ▼
Get log of recent invalidations (for debugging).
def get_invalidation_log() -> list[tuple[str, str, float]]
Returns
list[tuple[str, str, float]]
clear_invalidation_log 0 None ▼
Clear invalidation log.
def clear_invalidation_log() -> None