handler_list.py 2.33 KB
Newer Older
1 2
from typing import List, Tuple, Callable, Any
from yangson.context import Context
3
from yangson.schema import SchemaNode
4

5
HandlerSelectorT = Any
6

7 8

class BaseHandlerList:
9
    def __init__(self):
10
        self.handlers = []              # type: List[Tuple[HandlerSelectorT, Callable]]
11 12
        self.default_handler = None     # type: Callable

13
    def register_handler(self, identifier: str, handler: Callable):
14
        raise NotImplementedError("Not implemented in base class")
15 16 17 18

    def register_default_handler(self, handler: Callable):
        self.default_handler = handler

19
    def get_handler(self, identifier: str) -> Callable:
20 21 22 23
        raise NotImplementedError("Not implemented in base class")


class OpHandlerList(BaseHandlerList):
24 25
    def register_handler(self, op_name: str, handler: Callable):
        self.handlers.append((op_name, handler))
26

27
    def get_handler(self, op_name: str) -> Callable:
28
        for h in self.handlers:
29
            if h[0] == op_name:
30 31 32 33 34
                return h[1]

        return self.default_handler


35 36 37 38
class ConfDataHandlerList:
    def __init__(self):
        self.handlers = []  # type: List[Tuple[HandlerSelectorT, BaseDataListener]]

39
    def register_handler(self, handler: "BaseDataListener"):
40 41 42 43 44 45 46 47 48 49 50 51
        schema_node = handler.schema_node  # type: SchemaNode
        sch_node_id = str(id(schema_node))
        self.handlers.append((sch_node_id, handler))

    def get_handler(self, sch_node_id: str) -> Callable:
        for h in self.handlers:
            if h[0] == sch_node_id:
                return h[1]

        return None


52 53 54
class StateDataHandlerList:
    def __init__(self):
        self.handlers = []
55 56

    def register_handler(self, handler: "StateNodeHandlerBase"):
57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75
        saddr = Context.path2route(handler.sch_pth)
        self.handlers.append((saddr, handler))

    def get_handler(self, sch_pth: str, allow_superior: bool = True) -> Callable:
        saddr = Context.path2route(sch_pth)
        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


76
OP_HANDLERS = OpHandlerList()
77
STATE_DATA_HANDLES = StateDataHandlerList()
78
CONF_DATA_HANDLES = ConfDataHandlerList()