handler_list.py 1.92 KB
Newer Older
1
from typing import List, Dict, Tuple, Callable
2

3
from yangson.schemadata import SchemaData
4
from yangson.typealiases import SchemaRoute
5

6
from .handler_base import ConfDataHandlerBase, StateDataHandlerBase, ConfDataHandler, StateDataHandler, OpHandler
7 8


9
# ---------- Handler lists ----------
10 11
class ConfDataHandlerList:
    def __init__(self):
12 13
        self.handlers = {}  # type: Dict[int, ConfDataHandlerBase]
        self.handlers_pth = {}  # type: Dict[str, ConfDataHandlerBase]
14

15
    def register(self, handler: ConfDataHandlerBase):
16
        sch_node_id = id(handler.schema_node)
17 18
        self.handlers[sch_node_id] = handler
        self.handlers_pth[handler.schema_path] = handler
19

20
    def get_handler(self, sch_node_id: int) -> ConfDataHandler:
21
        return self.handlers.get(sch_node_id)
22

23
    def get_handler_by_pth(self, sch_pth: str) -> ConfDataHandler:
24
        return self.handlers_pth.get(sch_pth)
25 26


27 28
class StateDataHandlerList:
    def __init__(self):
29
        self.handlers = []  # type: List[Tuple[SchemaRoute, StateDataHandlerBase]]
30

31
    def register(self, handler: StateDataHandlerBase):
32
        saddr = SchemaData.path2route(handler.sch_pth)
33 34
        self.handlers.append((saddr, handler))

35
    def get_handler(self, sch_pth: str, allow_superior: bool = True) -> StateDataHandler:
36
        saddr = SchemaData.path2route(sch_pth)
37 38 39 40 41 42 43 44 45 46 47 48 49 50
        if allow_superior:
            while saddr:
                for h in self.handlers:
                    if h[0] == saddr:
                        return h[1]
                saddr.pop()
        else:
            for h in self.handlers:
                if h[0] == saddr:
                    return h[1]

        return None


51 52
class OpHandlerList:
    def __init__(self):
53
        self.handlers = {}  # type: Dict[str, OpHandler]
54

55 56 57
    def register(self, handler: Callable, op_name: str):
        self.handlers[op_name] = handler

58 59
    def get_handler(self, op_name: str) -> OpHandler:
        return self.handlers.get(op_name)