Add CacheDict.filepath and observe_access

exca/cachedict/__init__.py

@@ -6,6 +6,7 @@
 
 from .core import METADATA_TAG as METADATA_TAG
 from .core import CacheDict as CacheDict
+from .core import observe_access as observe_access
 
 CacheDictWriter = CacheDict  # deprecated: use CacheDict directly
 from exca.dumperloader import MEMMAP_ARRAY_FILE_MAX_CACHE as MEMMAP_ARRAY_FILE_MAX_CACHE

exca/cachedict/core.py

@@ -30,6 +30,49 @@
 logger = logging.getLogger(__name__)
 METADATA_TAG = "metadata="
 
+_access_observers: list[tp.Callable[[Path], None]] = []
+_observers_lock = threading.Lock()
+
+
+@contextlib.contextmanager
+def observe_access(
+    callback: tp.Callable[[Path], None],
+) -> tp.Iterator[tp.Callable[[Path], None]]:
+    """Call *callback* with the folder of every ``CacheDict`` touched in scope.
+
+    For the duration of the ``with`` block, *callback* receives the ``folder``
+    of each :class:`CacheDict` as it is constructed or accessed (``keys``,
+    ``__getitem__``, ``__setitem__``, ``__contains__``). Observation is
+    process-wide and nested scopes each get their own notifications.
+
+    Callbacks must be thread-safe (they may be invoked from cache reader
+    threads) and must not raise; exceptions are caught and logged at debug
+    level. Yields *callback* for ``with observe_access(cb):`` usage.
+    """
+    with _observers_lock:
+        _access_observers.append(callback)
+    try:
+        yield callback
+    finally:
+        with _observers_lock:
+            try:
+                _access_observers.remove(callback)
+            except ValueError:
+                pass
+
+
+def _notify_access(folder: Path | None) -> None:
+    """Notify active :func:`observe_access` callbacks of a touched *folder*."""
+    if folder is None or not _access_observers:
+        return
+    with _observers_lock:
+        observers = list(_access_observers)
+    for callback in observers:
+        try:
+            callback(folder)
+        except Exception:  # observation must never break caching
+            logger.debug("cache-access observer failed", exc_info=True)
+
 
 @dataclasses.dataclass
 class DumpInfo:
@@ -144,6 +187,7 @@ def __init__(
         if self.folder is not None:
             self._dumper = DumpContext(self.folder, permissions=self.permissions)
         self._local = threading.local()  # per-thread write context, see _write_ctx
+        _notify_access(self.folder)
 
     def __repr__(self) -> str:
         name = self.__class__.__name__
@@ -181,6 +225,7 @@ def __len__(self) -> int:
     def keys(self) -> tp.Iterator[str]:
         """Returns the keys in the dictionary
         (triggers a cache folder reading if folder is not None)"""
+        _notify_access(self.folder)
         self._read_info_files()
         keys = set(self._ram_data) | set(self._key_info)
         return iter(keys)
@@ -271,6 +316,7 @@ def items(self) -> tp.Iterator[tuple[str, X]]:
             yield key, self[key]
 
     def __getitem__(self, key: str) -> X:
+        _notify_access(self.folder)
         if self._keep_in_ram:
             if key in self._ram_data or self.folder is None:
                 return self._ram_data[key]
@@ -285,6 +331,37 @@ def __getitem__(self, key: str) -> X:
             self._ram_data[key] = loaded
         return loaded  # type: ignore
 
+    def filepath(self, key: str) -> Path:
+        """Absolute path of the file (or directory) backing *key* on disk.
+
+        Indexing returns the loaded value; this returns the path it was dumped
+        to instead (parquet/npy/pickle dumps, directory entries, ...). Triggers
+        a cache-folder read if *key* has not been loaded yet.
+
+        Raises
+        ------
+        RuntimeError
+            if this cache is RAM-only (``folder is None``).
+        KeyError
+            if *key* is not present in the cache.
+        ValueError
+            if *key*'s entry is not backed by a single file/directory (its
+            info record carries no ``filename``, e.g. an inline JSON value).
+        """
+        if self.folder is None:
+            raise RuntimeError("filepath requires a folder-backed CacheDict")
+        if key not in self._key_info:
+            _ = self.keys()  # reload keys from disk
+        if key not in self._key_info:
+            raise KeyError(key)
+        filename = self._key_info[key].content.get("filename")
+        if filename is None:
+            raise ValueError(
+                f"Entry {key!r} is not backed by a single file "
+                "(no 'filename' in its cache info)"
+            )
+        return (self.folder / filename).resolve()
+
     # Thread-local write context: each thread gets its own DumpContext
     # (and thus its own JSONL file), enabling concurrent writers.
     @property
@@ -325,6 +402,7 @@ def writer(self) -> tp.Iterator["CacheDict[X]"]:
             yield self
 
     def __setitem__(self, key: str, value: X) -> None:
+        _notify_access(self.folder)
         if not isinstance(key, str):
             raise TypeError(f"Non-string keys are not allowed (got {key!r})")
         if self.folder is not None and self._write_ctx is None:
@@ -368,6 +446,7 @@ def __delitem__(self, key: str) -> None:
             self._dumper.delete(info.content)
 
     def __contains__(self, key: str) -> bool:
+        _notify_access(self.folder)
         # in-memory cache
         if key in self._ram_data:
             return True

exca/cachedict/test_cachedict.py

@@ -391,3 +391,72 @@ def test_orphaned_cleanup_file_deleted_concurrently(tmp_path: Path) -> None:
         keys = list(cache.keys())
     assert keys == []
     assert reader._fp.name not in cache._jsonl_readers
+
+
+def test_filepath(tmp_path: Path) -> None:
+    cache: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=tmp_path, keep_in_ram=False)
+    with cache.write():
+        cache["arr"] = np.array([1, 2, 3])
+    fp = cache.filepath("arr")
+    assert fp.is_absolute()
+    assert fp.exists()
+    assert tmp_path.resolve() in fp.parents
+    # a fresh view (empty _key_info) resolves by reloading keys from disk
+    cache2: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=tmp_path)
+    assert cache2.filepath("arr") == fp
+    with pytest.raises(KeyError):
+        cache2.filepath("missing")
+
+
+def test_filepath_ram_only_raises() -> None:
+    cache: cd.CacheDict[int] = cd.CacheDict(folder=None, keep_in_ram=True)
+    with pytest.raises(RuntimeError):
+        cache.filepath("x")
+
+
+def test_filepath_inline_value_raises(tmp_path: Path) -> None:
+    cache: cd.CacheDict[tp.Any] = cd.CacheDict(folder=tmp_path, keep_in_ram=False)
+    # simulate an entry whose info record carries no 'filename' (inline value)
+    cache._key_info["x"] = cd.DumpInfo(
+        jsonl=tmp_path / "w-info.jsonl", byte_range=(0, 0), content={"#type": "Json"}
+    )
+    with pytest.raises(ValueError, match="not backed by a single file"):
+        cache.filepath("x")
+
+
+def test_observe_access(tmp_path: Path) -> None:
+    folder = tmp_path / "sub"
+    seen: list[Path] = []
+    with cd.observe_access(seen.append):
+        cache: cd.CacheDict[np.ndarray] = cd.CacheDict(folder=folder)  # __init__
+        with cache.write():
+            cache["a"] = np.array([1])  # __setitem__
+        assert "a" in cache  # __contains__
+        assert set(cache.keys()) == {"a"}  # keys
+        _ = cache["a"]  # __getitem__
+    # every notification is for this cache's folder, and construction was seen
+    assert seen  # got at least the __init__ notification
+    assert set(seen) == {folder}
+    # a RAM-only cache (folder is None) never notifies
+    seen.clear()
+    with cd.observe_access(seen.append):
+        cd.CacheDict(folder=None, keep_in_ram=True)
+    assert seen == []
+    # after every scope exits, the registry is empty again
+    assert cd._access_observers == []
+
+
+def test_observe_access_nested_and_error_safe(tmp_path: Path) -> None:
+    outer: list[Path] = []
+    inner: list[Path] = []
+
+    def boom(_folder: Path) -> None:
+        raise RuntimeError("observer errors must never break caching")
+
+    with cd.observe_access(outer.append):
+        with cd.observe_access(boom):  # raising observer is swallowed
+            with cd.observe_access(inner.append):
+                cd.CacheDict(folder=tmp_path / "x")
+    assert (tmp_path / "x") in outer
+    assert (tmp_path / "x") in inner
+    assert cd._access_observers == []