Add KVP helper and plugin file flags
Introduce a new python/helpers/kvp.py providing a thread-safe runtime and persistent key-value store (JSON files under usr/kvp) with key validation, find/get/set/remove operations, and atomic writes using temp files and os.replace. Update python/helpers/plugins.py to add has_readme, has_license, and has_init_script flags to PluginListItem and set them when scanning plugin directories (checks for README.md, LICENSE, and initialize.py).
frdel committed
Feb 25, 2026 at 11:38 UTC
6d4fd69570dee4df4804d65a56f1261c29f28a93
2 files changed
+118
python/helpers/kvp.py
new
+112
@@ -0,0 +1,112 @@
1
+import fnmatch
2
+import glob
3
+import json
4
+import os
5
+import tempfile
6
+import threading
7
+from typing import Any
8
+
9
+from python.helpers.files import get_abs_path
10
+
11
+_runtime_lock = threading.RLock()
12
+_runtime_store: dict[str, Any] = {}
13
+
14
+_persistent_lock = threading.RLock()
15
+
16
+
17
+def _persistent_dir() -> str:
18
+ return get_abs_path("usr", "kvp")
19
+
20
+
21
+def _validate_key(key: str) -> None:
22
+ if not key:
23
+ raise ValueError("key must not be empty")
24
+ if "\x00" in key:
25
+ raise ValueError("key contains NUL")
26
+ if "/" in key or os.path.sep in key or (os.path.altsep and os.path.altsep in key):
27
+ raise ValueError("key must not contain path separators")
28
+
29
+
30
+def _key_to_path(key: str) -> str:
31
+ _validate_key(key)
32
+ return os.path.join(_persistent_dir(), f"{key}.json")
33
+
34
+
35
+def get_runtime(key: str, default: Any = None) -> Any:
36
+ with _runtime_lock:
37
+ return _runtime_store.get(key, default)
38
+
39
+
40
+def set_runtime(key: str, value: Any) -> None:
41
+ _validate_key(key)
42
+ with _runtime_lock:
43
+ _runtime_store[key] = value
44
+
45
+
46
+def remove_runtime(key: str) -> None:
47
+ with _runtime_lock:
48
+ _runtime_store.pop(key, None)
49
+
50
+
51
+def find_runtime(pattern: str) -> list[str]:
52
+ if not pattern:
53
+ return []
54
+ with _runtime_lock:
55
+ return sorted([k for k in _runtime_store.keys() if fnmatch.fnmatch(k, pattern)])
56
+
57
+
58
+def get_persistent(key: str, default: Any = None) -> Any:
59
+ path = _key_to_path(key)
60
+ with _persistent_lock:
61
+ try:
62
+ with open(path, "r", encoding="utf-8") as f:
63
+ return json.load(f)
64
+ except FileNotFoundError:
65
+ return default
66
+
67
+
68
+def set_persistent(key: str, value: Any) -> None:
69
+ path = _key_to_path(key)
70
+ dir_path = os.path.dirname(path)
71
+
72
+ with _persistent_lock:
73
+ os.makedirs(dir_path, exist_ok=True)
74
+
75
+ fd, tmp_path = tempfile.mkstemp(prefix=f"{key}.", suffix=".tmp", dir=dir_path)
76
+ try:
77
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
78
+ json.dump(value, f, ensure_ascii=False, separators=(",", ":"))
79
+ f.flush()
80
+ os.fsync(f.fileno())
81
+ os.replace(tmp_path, path)
82
+ finally:
83
+ try:
84
+ if os.path.exists(tmp_path):
85
+ os.unlink(tmp_path)
86
+ except OSError:
87
+ pass
88
+
89
+
90
+def remove_persistent(key: str) -> None:
91
+ path = _key_to_path(key)
92
+ with _persistent_lock:
93
+ try:
94
+ os.unlink(path)
95
+ except FileNotFoundError:
96
+ return
97
+
98
+
99
+def find_persistent(pattern: str) -> list[str]:
100
+ if not pattern:
101
+ return []
102
+
103
+ dir_path = _persistent_dir()
104
+ with _persistent_lock:
105
+ if not os.path.isdir(dir_path):
106
+ return []
107
+
108
+ search = os.path.join(dir_path, f"{pattern}.json")
109
+ paths = glob.glob(search)
110
+ keys = [os.path.basename(p)[: -len(".json")] for p in paths]
111
+ keys.sort()
112
+ return keys
python/helpers/plugins.py
+6
@@ -52,6 +52,9 @@ class PluginListItem(BaseModel):
52
is_custom: bool = False
53
has_main_screen: bool = False
54
has_config_screen: bool = False
55
+ has_readme: bool = False
56
+ has_license: bool = False
57
+ has_init_script: bool = False
58
toggle_state: ToggleState = "disabled"
59
60
@@ -98,6 +101,9 @@ def get_enhanced_plugins_list(
101
)
102
has_main_screen = files.exists(str(d / "webui" / "main.html"))
103
has_config_screen = files.exists(str(d / "webui" / "config.html"))
104
+ has_readme = files.exists(str(d / "README.md"))
105
+ has_license = files.exists(str(d / "LICENSE"))
106
+ has_init_script = files.exists(str(d / "initialize.py"))
107
toggle_state = get_toggle_state(meta.name)
108
results.append(
109
PluginListItem(