manager: add a FunkyCache manager to track functions and dependencies
- ID
954bbfb- date
2026-11-06 13:53:28+00:00- author
Alex Chan <alex@alexwlchan.net>- parent
9e42abf- message
manager: add a FunkyCache manager to track functions and dependencies- changed files
2 files, 544 additions, 87 deletions
Changed files
src/funkycache/manager.py (4214 → 10189)
diff --git a/src/funkycache/manager.py b/src/funkycache/manager.py
index 70e375b..c4570fc 100644
--- a/src/funkycache/manager.py
+++ b/src/funkycache/manager.py
@@ -5,12 +5,17 @@ Ensure each function has an independent cache.
"""
import atexit
+from collections.abc import Callable
+from dataclasses import dataclass
from functools import wraps
import inspect
+import json
from pathlib import Path
-from typing import Literal, overload, ParamSpec, Protocol, TypeVar
+from typing import Any, Literal, overload, ParamSpec, Protocol, TypeVar
import warnings
+import xxhash
+
from .storage import (
CacheStorage,
MemoryCacheStorage,
@@ -37,17 +42,46 @@ class F[**P, R](Protocol):
StorageType = Literal["memory", "sqlite"]
+@dataclass
+class KnownCache[R]:
+ """
+ Tracks the underlying storage for a single function.
+ """
+
+ storage: CacheStorage
+ data_type: R
+
+ func_name: str
+ input_sig: str
+
+ dep_paths: set[Path]
+ dep_values: list[Any]
+
+ def is_affected_by(self, changed_paths: list[Path]) -> bool:
+ """
+ Whether this function depends on any of these paths.
+ """
+ for dp in self.dep_paths:
+ if dp in changed_paths:
+ return True
+ elif dp.is_dir() and any(p.is_relative_to(dp) for p in changed_paths):
+ return True
+ return False
+
+
class FunkyCache:
"""
FunkyCache manages all the cached functions in a codebase.
- Create a single instance of FunkyCache with your preferred settings --
- what sort of storage to use, and where to persist artefacts.
+ It tracks all of the functions being cached and the files/values
+ they depend on. When FunkyCache is notified that a file changes (or it
+ detects a change when the Python process is restarted), it invalidates
+ the cache.
"""
_storage: StorageType
_root: Path
- _known_caches: dict[str, CacheStorage]
+ _known_caches: dict[str, KnownCache]
@overload
def __init__(self, *, storage: Literal["memory"] = "memory"): ...
@@ -70,60 +104,95 @@ class FunkyCache:
self._root = Path(root)
self._known_caches = {}
- def register[**P, R](self, func: F) -> F:
+ @overload
+ def register[**P, R](self, func: F) -> F: ...
+
+ @overload
+ def register[**P, R](
+ self, *, files: list[Path | str] | None = None, values: list[Any] | None = None
+ ) -> Callable[[F], F]: ...
+
+ def register[**P, R](
+ self,
+ func: F | None = None,
+ *,
+ files: list[Path | str] | None = None,
+ values: list[Any] | None = None,
+ ) -> F:
"""
Cache the input/output of func.
- """
- sig = inspect.signature(func)
-
- func_name = self._canonical_spec(func)
- if func_name in self._known_caches:
- raise TypeError(f"Duplicate cache for {func_name}")
- storage = self._create_storage(func, data_type=sig.return_annotation)
- self._known_caches[func_name] = storage
-
- @wraps(func)
- def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
- bound = sig.bind(*args, **kwargs)
- bound.apply_defaults()
- key = bound.arguments
-
- try:
- return storage.get(key)
- except KeyError:
- pass
- except Exception as e: # pragma: no cover
- warnings.warn(f"unexpected error looking up key in cache: {e}")
-
- new_value = func(*args, **kwargs)
-
- try:
- storage.set(key, new_value)
- except CannotOverwriteValueException: # pragma: no cover
- raise
- except Exception as e: # pragma: no cover
- warnings.warn(f"unexpected error inserting value into cache: {e}")
- pass
-
- return new_value
-
- return wrapper
-
- @staticmethod
- def _canonical_spec(func: F) -> str:
- """
- Return a canonical identifier for this function.
- This should uniquely identify a function within a codebase,
- and functions with the same name but in different files/classes
- should get a different identifier.
+ The function can optionally declare a dependency on input files or
+ static values. If the files change, use notify_path_change to
+ invalidate functions that relied on that file. Static values are
+ presumed to be unchanged for the life of the Python process.
"""
- module = getattr(func, "__module__", "__main__")
- qualname = getattr(func, "__qualname__", func.__name__)
-
- return f"{module}::{qualname}"
- def _create_storage(self, func: F, data_type: R) -> CacheStorage[R]:
+ def decorator(func: F) -> F:
+ sig = inspect.signature(func)
+
+ dep_paths = _dep_paths(func, files=[Path(p) for p in (files or [])])
+ dep_values = values or []
+
+ func_name = _canonical_spec(func)
+ if func_name in self._known_caches:
+ raise TypeError(f"Duplicate cache for {func_name}")
+ input_sig = _input_signature(dep_paths, dep_values)
+ storage = self._create_storage(
+ func_name,
+ data_type=sig.return_annotation,
+ input_sig=input_sig,
+ )
+ self._known_caches[func_name] = KnownCache(
+ storage,
+ data_type=sig.return_annotation,
+ func_name=func_name,
+ input_sig=input_sig,
+ dep_paths=dep_paths,
+ dep_values=dep_values,
+ )
+
+ @wraps(func)
+ def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
+ bound = sig.bind(*args, **kwargs)
+ bound.apply_defaults()
+ key = bound.arguments
+
+ # We need to look up the CacheStorage in self._known_caches
+ # every time, in case it's changed since the last call.
+ storage = self._known_caches[func_name].storage
+
+ try:
+ return storage.get(key)
+ except KeyError:
+ pass
+ except Exception as e: # pragma: no cover
+ warnings.warn(f"unexpected error looking up key in cache: {e}")
+
+ new_value = func(*args, **kwargs)
+
+ try:
+ storage.set(key, new_value)
+ except CannotOverwriteValueException: # pragma: no cover
+ raise
+ except Exception as e: # pragma: no cover
+ warnings.warn(f"unexpected error inserting value into cache: {e}")
+ pass
+
+ return new_value
+
+ return wrapper
+
+ if func is not None:
+ return decorator(func)
+ return decorator
+
+ def _create_storage(
+ self,
+ func_name: str,
+ data_type: R,
+ input_sig: str,
+ ) -> CacheStorage[R]:
"""
Create a new instance of CacheStorage.
"""
@@ -131,12 +200,123 @@ class FunkyCache:
case "memory":
return MemoryCacheStorage[R]()
case "sqlite":
+ # Remove path-unsafe names from the function name,
+ # and use this as the on-disk identifier.
+ identifier = func_name.replace(":", "-")
+
+ # When we create a new SQLite cache, clear out any
+ # similar-looking caches which are presumably stale.
+ # TODO: Check if the cache was accessed recently, which
+ # might indicate another process still using it.
self._root.mkdir(exist_ok=True, parents=True)
- identifier = self._canonical_spec(func).replace(":", "-")
- storage = SQLiteCacheStorage[R](
- self._root / f"{identifier}.db", data_type=data_type
- )
+ db_path = self._root / f"{identifier}.{input_sig}.db"
+ storage = SQLiteCacheStorage[R](db_path, data_type=data_type)
+ for f in self._root.glob(f"{identifier}.*.db"):
+ if f != db_path:
+ f.unlink()
atexit.register(lambda: storage.close())
return storage
case _: # pragma: no cover
- raise TypeError(f"unrecognised storage: {storage!r}")
+ raise TypeError(f"unrecognised storage: {self._storage!r}")
+
+ def notify_path_change(self, changed_paths: Path | list[Path]) -> None:
+ """
+ Invalidate cache that depend on this path and/or paths.
+ """
+ if not isinstance(changed_paths, list):
+ changed_paths = [changed_paths]
+
+ for name, known_cache in self._known_caches.items():
+ # If this cache isn't affected by any of the paths that
+ # changed, we can leave it as-is.
+ if not known_cache.is_affected_by(changed_paths):
+ continue
+
+ # Compute the new input signature. If it's the same, we can
+ # keep the cache as-is.
+ new_input_sig = _input_signature(
+ known_cache.dep_paths, known_cache.dep_values
+ )
+
+ if known_cache.input_sig == new_input_sig:
+ continue
+
+ # This cache depends on a path that changed! Create a new cache.
+ if isinstance(known_cache.storage, SQLiteCacheStorage):
+ known_cache.storage.close()
+
+ known_cache.storage = self._create_storage(
+ func_name=known_cache.func_name,
+ data_type=known_cache.data_type,
+ input_sig=new_input_sig,
+ )
+ known_cache.input_sig = new_input_sig
+
+
+def _canonical_spec(func: F) -> str:
+ """
+ Return a canonical identifier for this function.
+
+ This should uniquely identify a function within a codebase,
+ and functions with the same name but in different files/classes
+ should get a different identifier.
+ """
+ module = getattr(func, "__module__", "__main__")
+ qualname = getattr(func, "__qualname__", func.__name__)
+
+ return f"{module}::{qualname}"
+
+
+def _dep_paths(func: F, files: list[Path]) -> set[Path]:
+ """
+ Get a list of all the paths this function depends on.
+ """
+ dep_paths: list[Path] = []
+
+ # 1. Include the source file of the function.
+ if func_file := inspect.getsourcefile(func):
+ dep_paths.append(Path(func_file))
+
+ # 2. Include any explicitly declared files.
+ if files:
+ dep_paths.extend(files)
+
+ return set(dep_paths)
+
+
+def _input_signature(dep_paths: set[Path], values: list[Any]) -> str:
+ """
+ Create a "signature" for all the inputs this cache depends on.
+
+ We get a list of all the files this function depends on, their hashes
+ (or "<missing>" if they don't exist yet), and the values. These get
+ saved in a JSON object to avoid collisions, and then we use the xxh64.
+ """
+ signature = {
+ "paths": {},
+ "values": values,
+ }
+
+ # For each dependent path, include both the path itself
+ # and a hash of the contents.
+ all_paths = []
+ for p in dep_paths:
+ if p.is_file() or not p.exists(follow_symlinks=False):
+ all_paths.append(p)
+ elif p.is_dir():
+ all_paths.extend(gp for gp in p.rglob("*") if gp.is_file())
+ else:
+ raise TypeError(f"Invalid path: {p}")
+
+ for p in sorted(all_paths):
+ try:
+ h = xxhash.xxh64()
+ with open(p, "rb") as infile:
+ while chunk := infile.read(8192):
+ h.update(chunk)
+ signature["paths"][str(p)] = h.hexdigest()
+ except FileNotFoundError:
+ signature["paths"][str(p)] = "<missing>"
+
+ json_str = json.dumps(signature, sort_keys=True)
+ return xxhash.xxh64_hexdigest(json_str.encode("utf8"))
tests/test_manager.py (4510 → 13569)
diff --git a/tests/test_manager.py b/tests/test_manager.py
index 54eda11..5492263 100644
--- a/tests/test_manager.py
+++ b/tests/test_manager.py
@@ -5,7 +5,10 @@ Tests for the FunkyCache cache manager.
from abc import ABC, abstractmethod
from collections.abc import Iterator
from collections import Counter
+import inspect
from pathlib import Path
+import subprocess
+import textwrap
import pytest
@@ -30,9 +33,7 @@ class FunkyCacheTestSuite(ABC):
When you register a function, it gets cached and only called once
per input.
"""
- cache = FunkyCache(storage="memory")
-
- # Count how many times the underlying foundation was called
+ # Count how many times the underlying function was called
calls = Counter[str]()
@cache.register
@@ -44,7 +45,7 @@ class FunkyCacheTestSuite(ABC):
assert initial("apple") == "A"
assert calls == {"apple": 1}
- # The second call is a cache hit; doesn't call the real function
+ # The second call is a cache hit; skips the real function
assert initial("apple") == "A"
assert calls == {"apple": 1}
@@ -99,6 +100,195 @@ class FunkyCacheTestSuite(ABC):
def double(n: int) -> int: # pragma: no cover
return n * 2
+ def test_cache_invalidated_on_path_change(
+ self, cache: FunkyCache, tmp_path: Path
+ ) -> None:
+ """
+ If a function depends on a path and the path changes, the cache
+ is invalidated.
+ """
+ p = tmp_path / "greeting.txt"
+ p.write_text("hello world!")
+
+ # Count how many times the underlying function was called
+ calls = Counter[str]()
+
+ @cache.register(files=[p])
+ def initial(s: str) -> str:
+ calls[s] += 1
+ return s[0].upper()
+
+ # The first call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # The second call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # Change the underlying file
+ p.write_text("bonjour monde")
+ cache.notify_path_change([p])
+
+ # The third call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 2}
+
+ def test_cache_not_invalidated_if_path_change_is_noop(
+ self, cache: FunkyCache, tmp_path: Path
+ ) -> None:
+ """
+ If a function depends on a path and we get a notification that
+ the path changed, but it hasn't actually changed, we preserve
+ the cache.
+ """
+ p = tmp_path / "greeting.txt"
+ p.write_text("hello world!")
+
+ # Count how many times the underlying function was called
+ calls = Counter[str]()
+
+ @cache.register(files=[p])
+ def initial(s: str) -> str:
+ calls[s] += 1
+ return s[0].upper()
+
+ # The first call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # The second call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # Notify of a file change
+ cache.notify_path_change(p)
+
+ # The third call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ def test_depends_on_folder(self, cache: FunkyCache, tmp_path: Path) -> None:
+ """
+ If a function depends on a folder and a path under the folder
+ changes, the cache is invalided.
+ """
+ p = tmp_path / "greeting.txt"
+ p.write_text("hello world!")
+
+ # Count how many times the underlying function was called
+ calls = Counter[str]()
+
+ @cache.register(files=[tmp_path])
+ def initial(s: str) -> str:
+ calls[s] += 1
+ return s[0].upper()
+
+ # The first call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # The second call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # Change the underlying file, and send a notification for the
+ # directory changing
+ p.write_text("bonjour monde")
+ cache.notify_path_change(tmp_path)
+
+ # The third call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 2}
+
+ # Change the underlying file, and send a notification for the
+ # file changing
+ p.write_text("hallo monde")
+ cache.notify_path_change(p)
+
+ # The fourth call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 3}
+
+ def test_unrelated_change_is_ignored(
+ self, cache: FunkyCache, tmp_path: Path
+ ) -> None:
+ """
+ If a path changes but a function doesn't depend on it, the cache
+ is left as-is.
+ """
+ p = tmp_path / "greeting.txt"
+ p.write_text("hello world!")
+
+ # Count how many times the underlying function was called
+ calls = Counter[str]()
+
+ @cache.register
+ def initial(s: str) -> str:
+ calls[s] += 1
+ return s[0].upper()
+
+ # The first call is a cache miss; goes to the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # The second call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # Notify the cache of a changed path
+ cache.notify_path_change(tmp_path)
+
+ # The third call is a cache hit; skips the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ def test_skips_function_with_no_source_file(self, cache: FunkyCache) -> None:
+ """
+ If a function doesn't have a discoverable source file, it can
+ still be registered for caching.
+ """
+ code_obj = compile(
+ "def double(x: int) -> int: return x * 2", "<string>", "exec"
+ )
+ namespace = {}
+ exec(code_obj, namespace)
+ double = namespace["double"]
+
+ assert inspect.getsourcefile(double) is None
+
+ cache.register(double)
+
+ def test_function_cannot_depend_on_symlink(
+ self, cache: FunkyCache, tmp_path: Path
+ ) -> None:
+ """
+ A function cannot depend on a symlink.
+ """
+ p = tmp_path / "symlink"
+ p.symlink_to(tmp_path / "greeting.txt")
+
+ with pytest.raises(TypeError, match="Invalid path"):
+
+ @cache.register(files=[p])
+ def double(n: int) -> int: # pragma: no cover
+ return n * 2
+
+ def test_function_can_depend_on_missing_file(
+ self, cache: FunkyCache, tmp_path: Path
+ ) -> None:
+ """
+ A function can depend on a missing file. This allows a function
+ to register an interest in a file that doesn't exist yet, but
+ will be created later.
+ """
+
+ @cache.register(files=[tmp_path / "does_not_exist"])
+ def double(n: int) -> int:
+ return n * 2
+
+ assert double(5) == 10
+
class TestMemoryFunkyCache(FunkyCacheTestSuite):
"""
@@ -112,6 +302,15 @@ class TestMemoryFunkyCache(FunkyCacheTestSuite):
"""
yield FunkyCache(storage="memory")
+ def test_no_root_with_memory(self, tmp_path: Path) -> None:
+ """
+ You cannot construct an in-memory FunkyCache with a root.
+ """
+ with pytest.raises(
+ TypeError, match="Do not specify a root path with memory storage"
+ ):
+ FunkyCache(storage="memory", root=tmp_path / ".cache") # type: ignore
+
class TestSQLiteFunkyCache(FunkyCacheTestSuite):
"""
@@ -125,36 +324,114 @@ class TestSQLiteFunkyCache(FunkyCacheTestSuite):
"""
yield FunkyCache(storage="sqlite", root=tmp_path / ".cache")
+ def test_cache_varies_based_on_static_values(self, tmp_path: Path) -> None:
+ """
+ The static values that a function depends on are used to distinguish
+ cache instances.
+ """
+ c1 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
-def test_sqlite_cache(tmp_path: Path) -> None:
- """
- Data can be persisted to an SQLite cache on-disk.
- """
- # Create a new cache, and populate it with some data
- cache1 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
+ @c1.register(values=[1])
+ def myfunc() -> str:
+ return "the value is one"
- @cache1.register
- def initial(s: str) -> str:
- return s[0].upper()
+ assert myfunc() == "the value is one"
- assert initial("apple") == "A"
+ c2 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
- # Open a new instance of FunkyCache pointing at the same cache root,
- # and check it retrieves the cached value.
- cache2 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
+ @c2.register(values=[2])
+ def myfunc() -> str:
+ return "the value is two"
- @cache2.register
- def initial(s: str) -> str: # pragma: no cover
- assert 0, "should not be called"
+ assert myfunc() == "the value is two"
- assert initial("apple") == "A"
+ def test_stale_caches_are_removed(self, tmp_path: Path) -> None:
+ """
+ If the cache for a function is invalidated, the old SQLite
+ database gets cleaned up.
+ """
+ root = tmp_path / ".cache"
+ c = FunkyCache(storage="sqlite", root=root)
+ @c.register(files=[tmp_path / "greeting.txt"])
+ def double(n: int) -> int:
+ return n * 2
-def test_no_root_with_memory(tmp_path: Path) -> None:
- """
- You cannot construct an in-memory FunkyCache with a root.
- """
- with pytest.raises(
- TypeError, match="Do not specify a root path with memory storage"
- ):
- FunkyCache(storage="memory", root=tmp_path / ".cache") # type: ignore
+ assert double(5) == 10
+
+ # After populating the cache, there's a single set of SQLite files
+ # in the .cache directory
+ stem1 = {p.stem for p in root.iterdir()}
+ assert len(stem1) == 1
+
+ # Invalidate the cache
+ p = tmp_path / "greeting.txt"
+ p.write_text("hello world")
+ c.notify_path_change(p)
+
+ # There should be a new set of SQLite files in the .cache directory
+ stem2 = {p.stem for p in root.iterdir()}
+ assert len(stem2) == 1
+ assert stem1 != stem2
+
+ def test_cache_depends_on_python_file(self, tmp_path: Path) -> None:
+ """
+ The cache depends on the source file of the cached function.
+ """
+ p = tmp_path / "example.py"
+
+ def run():
+ cmd = ["python3", str(p)]
+ return subprocess.check_output(cmd, cwd=tmp_path, text=True).strip()
+
+ # Write the initial version of the file, and check the first call
+ # is a cache miss, subsequent calls are cache hits
+ p.write_text(
+ textwrap.dedent("""
+ from funkycache import FunkyCache
+
+ GREETING = "hello world"
+
+ cache = FunkyCache(storage="sqlite", root=".cache")
+
+ @cache.register
+ def greet() -> str:
+ print("calling original function")
+ return GREETING
+
+ print(greet())
+""")
+ )
+
+ assert run() == "calling original function\nhello world"
+ assert run() == "hello world"
+ assert run() == "hello world"
+
+ # Update the file, and check the next call is a cache miss
+ new_text = p.read_text().replace('"hello world"', '"bonjour monde"')
+ p.write_text(new_text)
+
+ assert run() == "calling original function\nbonjour monde"
+
+ def test_data_is_persistent(self, tmp_path: Path) -> None:
+ """
+ Data can be persisted to an SQLite cache on-disk.
+ """
+ # Create a new cache, and populate it with some data
+ cache1 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
+
+ @cache1.register
+ def initial(s: str) -> str:
+ return s[0].upper()
+
+ assert initial("apple") == "A"
+
+ # Open a new instance of FunkyCache pointing at the same cache root,
+ # and check it retrieves the cached value.
+ cache2 = FunkyCache(storage="sqlite", root=tmp_path / ".cache")
+
+ @cache2.register
+ def initial(s: str) -> str: # pragma: no cover
+ assert 0, "should not be called"
+
+ assert initial("apple") == "A"