commonwealth: decorators: add cache invalidation

Author: nicoschmdt

core/libs/commonwealth/src/commonwealth/utils/decorators.py

@@ -9,6 +9,10 @@
 def temporary_cache(timeout_seconds: float = 10) -> Callable[[F], F]:
     """Decorator that creates a cache for specific inputs with a configured timeout in seconds.
 
+    The wrapped function exposes an `invalidate()` attribute that drops every cached entry,
+    forcing the next call to re-execute the function. Useful when the cached data is known
+    to be stale (e.g. after a write that mutates the underlying state).
+
     Args:
         timeout_seconds (float, optional): Timeout to be used for cache invalidation. Defaults to 10.
 
@@ -35,6 +39,11 @@ def wrapper(*args: Any) -> Any:
             cache[args] = function_return
             return function_return
 
+        def invalidate() -> None:
+            cache.clear()
+            last_sample_time.clear()
+
+        wrapper.invalidate = invalidate  # type: ignore[attr-defined]
         return wrapper  # type: ignore
 
     return inner_function

core/libs/commonwealth/src/commonwealth/utils/tests/test_decorators.py

@@ -24,3 +24,16 @@ def test_nested_settings_save_load() -> None:
 
     # Check if all cache values are invalid after waiting for a long time
     assert all(original_output[key] != cached_function(key) for key in inputs)
+
+
+def test_temporary_cache_invalidate() -> None:
+    inputs = ["first", "second", "third"]
+    original_output = {key: cached_function(key) for key in inputs}
+
+    # Cache is still warm: same call returns the same value
+    assert all(original_output[key] == cached_function(key) for key in inputs)
+
+    # Force invalidation; subsequent calls must re-execute the function
+    cached_function.invalidate()  # type: ignore[attr-defined]
+
+    assert all(original_output[key] != cached_function(key) for key in inputs)