aboutsummaryrefslogtreecommitdiffstats
path: root/deckit/module.py
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-11-09 18:11:41 +0100
committerDanilo M. <danix@danix.xyz>2026-11-09 18:11:41 +0100
commit5332d87511a9336c004af755f007aaa688d9c654 (patch)
treedf0a61dd62f303adbb425fd76896845fa964ca48 /deckit/module.py
parentd1afaffd414fcb9328f17983540076b539456bd3 (diff)
downloaddeck-it-5332d87511a9336c004af755f007aaa688d9c654.tar.gz
deck-it-5332d87511a9336c004af755f007aaa688d9c654.zip
feat: module contract and registry
Diffstat (limited to 'deckit/module.py')
-rw-r--r--deckit/module.py45
1 files changed, 45 insertions, 0 deletions
diff --git a/deckit/module.py b/deckit/module.py
new file mode 100644
index 0000000..b30fbf5
--- /dev/null
+++ b/deckit/module.py
@@ -0,0 +1,45 @@
+"""The module contract and the registry that dispatches to modules.
+
+A module is a data source plus an allow-listed set of actions. The client sends
+a module name and an action id; it never sends a command. The registry is the
+single place that maps ids to code, so adding a module never touches the server.
+"""
+
+from dataclasses import dataclass, field
+from typing import Callable
+
+
+class ModuleError(Exception):
+ """A bad module/action id, or a module that cannot serve the request."""
+
+
+@dataclass
+class Module:
+ name: str
+ title: str
+ state: Callable[[], dict]
+ actions: dict[str, Callable[[dict], dict | None]] = field(default_factory=dict)
+ start: Callable[[], None] | None = None
+ stop: Callable[[], None] | None = None
+
+
+class Registry:
+ def __init__(self, modules):
+ self._modules = {m.name: m for m in modules}
+
+ def names(self) -> list[dict]:
+ return [{"name": m.name, "title": m.title} for m in self._modules.values()]
+
+ def _get(self, name: str) -> Module:
+ if name not in self._modules:
+ raise ModuleError(f"unknown module: {name}")
+ return self._modules[name]
+
+ def state(self, name: str) -> dict:
+ return self._get(name).state()
+
+ def run_action(self, name: str, action: str, body: dict) -> dict | None:
+ module = self._get(name)
+ if action not in module.actions:
+ raise ModuleError(f"unknown action: {name}/{action}")
+ return module.actions[action](body)