jmri_server.service

 1"""``JsonService`` — base class for one topic's request handlers.
 2
 3Each topic (``turnout``, ``sensor``, ``power``, …) is one subclass. The
 4dispatcher resolves the topic from the request envelope's ``type`` and
 5calls one of :meth:`on_get`, :meth:`on_list`, :meth:`on_put`,
 6:meth:`on_post`, :meth:`on_delete` based on the method.
 7
 8Sub-classes return Python dicts; the dispatcher wraps them in the response
 9envelope. Errors are raised as :class:`JsonServiceError`; the dispatcher
10converts them to the JMRI error response format.
11
12Role gating (when ``JMRI_AUTH_ENABLED=1``): every method has a minimum
13role required to invoke it. Defaults live in :attr:`REQUIRED_ROLE` —
14``viewer`` for reads (``get`` / ``list``), ``operator`` for mutations
15(``post`` / ``put`` / ``delete``). Individual services override by
16setting :attr:`REQUIRED_ROLE` with their own per-method map — e.g.
17``configProfile`` requires ``admin`` for everything. The dispatcher
18enforces the check before invoking the handler; a role-gated failure
19raises :class:`JsonServiceError` with code 403.
20"""
21from __future__ import annotations
22
23from typing import TYPE_CHECKING, Any
24
25from jmri_server.auth.models import Role
26
27if TYPE_CHECKING:
28    from jmri_server.context import ServerContext
29    from jmri_server.envelope import Method
30
31
32class JsonServiceError(Exception):
33    """Raised by a service to produce a JMRI-format error response."""
34
35    def __init__(self, code: int, message: str) -> None:
36        super().__init__(message)
37        self.code = code
38        self.message = message
39
40
41class JsonService:
42    """Base class for a topic's service handlers.
43
44    Subclasses set :attr:`topic` and override the methods they support;
45    unsupported methods raise 405 by default. Subclasses can also
46    override :attr:`REQUIRED_ROLE` to change per-method role gates.
47    """
48
49    topic: str = ""  # subclass must set
50
51    # Minimum role required to invoke each method. Defaults: reads
52    # are ``viewer``, mutations are ``operator``. Override on a
53    # per-service basis — the class-level dict is merged with the
54    # base defaults so a subclass only lists the methods it changes.
55    REQUIRED_ROLE: dict[str, Role] = {
56        "get": Role.VIEWER,
57        "list": Role.VIEWER,
58        "post": Role.OPERATOR,
59        "put": Role.OPERATOR,
60        "delete": Role.OPERATOR,
61    }
62
63    def required_role(self, method: str) -> Role:
64        """Look up the role required to invoke *method* on this
65        service, defaulting to :data:`Role.OPERATOR` for unknown
66        methods (defense against a future custom verb landing
67        without a role declaration)."""
68        return self.REQUIRED_ROLE.get(method, Role.OPERATOR)
69
70    def __init__(self, ctx: ServerContext) -> None:
71        if not self.topic:
72            raise TypeError(f"{type(self).__name__}.topic must be set")
73        self.ctx = ctx
74
75    # Each method receives the request `data` dict (may be None) and returns
76    # either a dict (single-object reply) or a list of dicts (list reply).
77    Payload = dict[str, Any] | list[dict[str, Any]]
78
79    async def on_get(self, data: dict[str, Any] | None) -> Payload:
80        raise JsonServiceError(405, f"{self.topic!r} does not support method 'get'")
81
82    async def on_list(self, data: dict[str, Any] | None) -> Payload:
83        raise JsonServiceError(405, f"{self.topic!r} does not support method 'list'")
84
85    async def on_put(self, data: dict[str, Any] | None) -> Payload:
86        raise JsonServiceError(405, f"{self.topic!r} does not support method 'put'")
87
88    async def on_post(self, data: dict[str, Any] | None) -> Payload:
89        raise JsonServiceError(405, f"{self.topic!r} does not support method 'post'")
90
91    async def on_delete(self, data: dict[str, Any] | None) -> Payload:
92        raise JsonServiceError(405, f"{self.topic!r} does not support method 'delete'")