manager: create the initial manager class to track cache instances
- ID
e097186- date
2026-11-06 08:11:40+00:00- author
Alex Chan <alex@alexwlchan.net>- parent
280becb- message
manager: create the initial manager class to track cache instances- changed files
3 files, 252 additions, 1 deletion
Changed files
src/funkycache/__init__.py (219 → 249)
diff --git a/src/funkycache/__init__.py b/src/funkycache/__init__.py
index f904e09..954332b 100644
--- a/src/funkycache/__init__.py
+++ b/src/funkycache/__init__.py
@@ -2,7 +2,8 @@
funkycache is a SQLite-backed function cache that tracks file dependencies.
"""
+from .manager import FunkyCache
from .storage import CannotOverwriteValueException
__version__ = "0.0.1"
-__all__ = ["CacheStorage", "CannotOverwriteValueException"]
+__all__ = ["CannotOverwriteValueException", "FunkyCache"]
src/funkycache/manager.py (0 → 3379)
diff --git a/src/funkycache/manager.py b/src/funkycache/manager.py
new file mode 100644
index 0000000..e1740f2
--- /dev/null
+++ b/src/funkycache/manager.py
@@ -0,0 +1,121 @@
+"""
+Manage the caching of all the functions in a codebase.
+
+Ensure each function has an independent cache.
+"""
+
+import atexit
+from functools import wraps
+import inspect
+from pathlib import Path
+from typing import Literal, overload, ParamSpec, Protocol, TypeVar
+import warnings
+
+from .storage import (
+ CacheStorage,
+ MemoryCacheStorage,
+ SQLiteCacheStorage,
+ CannotOverwriteValueException,
+)
+
+
+P = ParamSpec("P")
+R = TypeVar("R")
+
+
+class F[**P, R](Protocol):
+ """
+ A function: that is, a callable that also exposes a __name__ attribute.
+ """
+
+ def __call__(self, *args: P.args, **kwargs: P.kwargs) -> R: ... # noqa: D102
+
+ @property
+ def __name__(self) -> str: ... # noqa: D105
+
+
+StorageType = Literal["memory", "sqlite"]
+
+
+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.
+ """
+
+ _storage: StorageType
+ _root: Path
+
+ @overload
+ def __init__(self, *, storage: Literal["memory"] = "memory"): ...
+
+ @overload
+ def __init__(
+ self, *, storage: Literal["sqlite"], root: Path | str | None = None
+ ) -> None: ...
+
+ def __init__(
+ self, *, storage: StorageType = "memory", root: Path | str | None = None
+ ):
+ if storage == "memory" and root is not None:
+ raise TypeError("Do not specify a root path with memory storage")
+
+ if root is None:
+ root = Path(".funkycache")
+
+ self._storage = storage
+ self._root = Path(root)
+
+ def register[**P, R](self, func: F) -> F:
+ """
+ Cache the input/output of func.
+ """
+ sig = inspect.signature(func)
+
+ storage = self._create_storage(func, data_type=sig.return_annotation)
+
+ @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
+
+ def _create_storage(self, func: F, data_type: R) -> CacheStorage[R]:
+ """
+ Create a new instance of CacheStorage.
+ """
+ match self._storage:
+ case "memory":
+ return MemoryCacheStorage[R]()
+ case "sqlite":
+ self._root.mkdir(exist_ok=True, parents=True)
+ storage = SQLiteCacheStorage[R](
+ self._root / f"{func.__name__}.db", data_type=data_type
+ )
+ atexit.register(lambda: storage.close())
+ return storage
+ case _: # pragma: no cover
+ raise TypeError(f"unrecognised storage: {storage!r}")
tests/test_manager.py (0 → 3572)
diff --git a/tests/test_manager.py b/tests/test_manager.py
new file mode 100644
index 0000000..9022e54
--- /dev/null
+++ b/tests/test_manager.py
@@ -0,0 +1,129 @@
+"""
+Tests for the FunkyCache cache manager.
+"""
+
+from abc import ABC, abstractmethod
+from collections.abc import Iterator
+from collections import Counter
+from pathlib import Path
+
+import pytest
+
+from funkycache import FunkyCache
+
+
+class FunkyCacheTestSuite(ABC):
+ """
+ Tests for the properties of FunkyCache which should be independent
+ of the backing storage.
+ """
+
+ @abstractmethod
+ @pytest.fixture
+ def cache(self, tmp_path: Path) -> Iterator[FunkyCache]:
+ """
+ Return an empty instance of FunkyCache.
+ """
+
+ def test_registered_function_is_cached(self, cache: FunkyCache) -> None:
+ """
+ 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
+ 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; doesn't call the real function
+ assert initial("apple") == "A"
+ assert calls == {"apple": 1}
+
+ # Calling with a different input is a cache miss; goes to the real function
+ assert initial("banana") == "B"
+ assert calls == {"apple": 1, "banana": 1}
+
+ def test_per_function_caches_are_independent(self, cache: FunkyCache) -> None:
+ """
+ Different functions have different caches.
+ """
+
+ @cache.register
+ def double(n: int) -> int:
+ return n * 2
+
+ @cache.register
+ def triple(n: int) -> int:
+ return n * 3
+
+ assert double(2) == 4
+ assert triple(2) == 6
+
+
+class TestMemoryFunkyCache(FunkyCacheTestSuite):
+ """
+ Tests for a FunkyCache which is backed by in-memory storage.
+ """
+
+ @pytest.fixture
+ def cache(self, tmp_path: Path) -> Iterator[FunkyCache]:
+ """
+ Return an empty instance of FunkyCache.
+ """
+ yield FunkyCache(storage="memory")
+
+
+class TestSQLiteFunkyCache(FunkyCacheTestSuite):
+ """
+ Tests for a FunkyCache which is backed by SQLite storage.
+ """
+
+ @pytest.fixture
+ def cache(self, tmp_path: Path) -> Iterator[FunkyCache]:
+ """
+ Return an empty instance of FunkyCache.
+ """
+ yield 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")
+
+ @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"
+
+
+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