1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
"""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]
_get = get
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)
|
