Dev

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v2.2.0 (2025-12-12)
+
+### Feat
+
+- **cache**: add cache utilities and tests
+
 ## v2.1.0 (2025-12-01)
 
 ### Feat

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "funcnodes-core"
 
-version = "2.1.0"
+version = "2.2.0"
 
 description = "core package for funcnodes"
 authors = [{name = "Julian Kimmig", email = "julian.kimmig@linkdlab.de"}]

src/funcnodes_core/utils/cache.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import shutil
+from pathlib import Path
+from typing import Any, Optional
+
+from .. import config as fnconfig
+from .files import write_json_secure
+
+
+def get_cache_dir(subdir: str = "cache") -> Path:
+    """
+    Return (and ensure) a cache directory under the funcnodes config dir.
+
+    Args:
+        subdir: Subdirectory name under the config dir.
+    """
+    cache_dir = fnconfig.get_config_dir() / subdir
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    return cache_dir
+
+
+def get_cache_path(filename: str, subdir: str = "cache") -> Path:
+    """Return full path for a cache file within the cache dir."""
+    return get_cache_dir(subdir) / filename
+
+
+def get_cache_meta_path_for(cache_path: Path) -> Path:
+    """Return a sidecar meta file path next to a cache file."""
+    return cache_path.with_suffix(cache_path.suffix + ".meta.json")
+
+
+def get_cache_meta_for(cache_path: Path) -> Optional[dict[str, Any]]:
+    """Read JSON metadata for a given cache file, if present."""
+    meta_path = get_cache_meta_path_for(cache_path)
+    if not meta_path.exists():
+        return None
+    try:
+        return json.loads(meta_path.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+
+
+def set_cache_meta_for(cache_path: Path, meta: dict[str, Any]):
+    """Write JSON metadata for a given cache file (atomic)."""
+    meta_path = get_cache_meta_path_for(cache_path)
+    write_json_secure(meta, meta_path, indent=2)
+
+
+def write_cache_text(cache_path: Path, text: str, encoding: str = "utf-8"):
+    """
+    Write text to cache_path using a temp file + atomic replace.
+
+    This avoids partially-written cache files if the process crashes mid-write.
+    """
+    cache_path = Path(cache_path)
+    cache_path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(
+        "w+", dir=cache_path.parent, delete=False, encoding=encoding
+    ) as temp_file:
+        temp_file.write(text)
+        temp_file.flush()
+        os.fsync(temp_file.fileno())
+        temp_path = temp_file.name
+    os.replace(temp_path, str(cache_path))
+
+
+def clear_cache():
+    """Clear all cache files."""
+    cache_dir = get_cache_dir()
+    shutil.rmtree(cache_dir, ignore_errors=True)
+    cache_dir.mkdir(parents=True, exist_ok=True)

tests/test_cache_utils.py

@@ -0,0 +1,78 @@
+import funcnodes_core as fn
+from pytest_funcnodes import funcnodes_test
+
+
+@funcnodes_test
+def test_get_cache_dir_creates_under_config_dir():
+    from funcnodes_core.utils.cache import get_cache_dir
+
+    cache_dir = get_cache_dir()
+    assert cache_dir.exists()
+    assert cache_dir.is_dir()
+    assert fn.config.get_config_dir() in cache_dir.parents
+
+
+@funcnodes_test
+def test_cache_meta_roundtrip():
+    from funcnodes_core.utils.cache import (
+        get_cache_path,
+        get_cache_meta_for,
+        set_cache_meta_for,
+    )
+
+    cache_path = get_cache_path("example.txt")
+    meta = {"foo": "bar", "num": 1}
+    set_cache_meta_for(cache_path, meta)
+
+    loaded = get_cache_meta_for(cache_path)
+    assert loaded == meta
+
+
+@funcnodes_test
+def test_write_cache_text_writes_file():
+    from funcnodes_core.utils.cache import get_cache_path, write_cache_text
+
+    cache_path = get_cache_path("example.txt")
+    write_cache_text(cache_path, "hello")
+
+    assert cache_path.exists()
+    assert cache_path.read_text(encoding="utf-8") == "hello"
+
+
+@funcnodes_test
+def test_clear_cache_clears_cache():
+    from funcnodes_core.utils.cache import clear_cache, get_cache_dir, write_cache_text
+
+    # check that it is empty
+    assert len(list(get_cache_dir().glob("*"))) == 0
+    # write a file to the cache
+    write_cache_text(get_cache_dir() / "test.txt", "hello")
+    assert (get_cache_dir() / "test.txt").exists()
+    # check that it is not empty
+    assert len(list(get_cache_dir().glob("*"))) == 1
+    # clear the cache
+    clear_cache()
+
+    # check that it is empty
+    assert get_cache_dir().exists()
+    assert len(list(get_cache_dir().glob("*"))) == 0
+
+
+@funcnodes_test
+def test_chache_meta_exception_handling():
+    from funcnodes_core.utils.cache import (
+        get_cache_path,
+        get_cache_meta_for,
+        set_cache_meta_for,
+    )
+
+    cache_path = get_cache_path("example.txt")
+    set_cache_meta_for(cache_path, "hello")
+
+    assert get_cache_meta_for(cache_path) == "hello"
+
+    # write a invalid meta file
+    (cache_path.with_suffix(cache_path.suffix + ".meta.json")).write_text("invalid")
+    assert get_cache_meta_for(cache_path) is None
+
+    assert get_cache_meta_for(cache_path.with_suffix(cache_path.suffix + ".md")) is None