File: dispatcher.py

package info (click to toggle)
python-ajsonrpc 1.2.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 524 kB
  • sloc: python: 1,286; makefile: 56; sh: 17
file content (164 lines) | stat: -rw-r--r-- 4,995 bytes parent folder | download
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
"""Method name to method mapper.

Dispatcher is a dict-like object which maps method_name to method.
For usage examples see :meth:`~Dispatcher.add_function`

"""
import functools
import inspect
import types
from typing import Any, Optional, Mapping
from collections.abc import Mapping as CollectionsMapping, MutableMapping, Callable


class Dispatcher(MutableMapping):

    """Dictionary-like object which maps method_name to method."""

    def __init__(self, prototype: Any = None, prefix: Optional[str] = None) -> None:
        """ Build method dispatcher.

        Parameters
        ----------
        prototype : object or dict, optional
            Initial method mapping.

        Examples
        --------

        Init object with method dictionary.

        >>> Dispatcher({"sum": lambda a, b: a + b})
        None

        """
        self.method_map: Mapping[str, Callable] = dict()

        if prototype is not None:
            self.add_prototype(prototype, prefix=prefix)

    def __getitem__(self, key: str) -> Callable:
        return self.method_map[key]

    def __setitem__(self, key: str, value: Callable) -> None:
        self.method_map[key] = value

    def __delitem__(self, key: str) -> None:
        del self.method_map[key]

    def __len__(self):
        return len(self.method_map)

    def __iter__(self):
        return iter(self.method_map)

    def __repr__(self):
        return repr(self.method_map)

    @staticmethod
    def _getattr_function(prototype: Any, attr: str) -> Callable:
        """Fix the issue of accessing instance method of a class.
        
        Class.method(self, *args **kwargs) requires the first argument to be
        instance, but it was not given. Substitute method with a partial
        function where the first argument is an empty class constructor.
        
        """

        method = getattr(prototype, attr)
        if inspect.isclass(prototype) and isinstance(prototype.__dict__[attr], types.FunctionType):
            return functools.partial(method, prototype())
        return method

    @staticmethod
    def _extract_methods(prototype: Any, prefix: str = "") -> Mapping[str, Callable]:
        return {
            prefix + attr: Dispatcher._getattr_function(prototype, attr)
            for attr in dir(prototype)
            if not attr.startswith("_")
        }

    def add_class(self, cls: Any, prefix: Optional[str] = None) -> None:
        """Add class to dispatcher.

        Adds all of the public methods to dispatcher.

        Notes
        -----
            If class has instance methods (e.g. no @classmethod decorator),
            they likely would not work. Use :meth:`~add_object` instead.
            At the moment, dispatcher creates an object with empty constructor
            for instance methods.

        Parameters
        ----------
        cls : type
            class with methods to be added to dispatcher
        prefix : str, optional
            Method prefix. If not present, lowercased class name is used.

        """
        if prefix is None:
            prefix = cls.__name__.lower() + '.'

        self.update(Dispatcher._extract_methods(cls, prefix=prefix))

    def add_object(self, obj: Any, prefix: Optional[str] = None) -> None:
        if prefix is None:
            prefix = obj.__class__.__name__.lower() + '.'

        self.update(Dispatcher._extract_methods(obj, prefix=prefix))

    def add_prototype(self, prototype: Any, prefix: Optional[str] = None) -> None:
        if isinstance(prototype, CollectionsMapping):
            self.update({
                (prefix or "") + key: value
                for key, value in prototype.items()
            })
        elif inspect.isclass(prototype):
            self.add_class(prototype, prefix=prefix)
        else:
            self.add_object(prototype, prefix=prefix)

    def add_function(self, f: Callable = None, name: Optional[str] = None) -> Callable:
        """ Add a method to the dispatcher.

        Parameters
        ----------
        f : callable
            Callable to be added.
        name : str, optional
            Name to register (the default is function **f** name)

        Notes
        -----
        When used as a decorator keeps callable object unmodified.

        Examples
        --------

        Use as method

        >>> d = Dispatcher()
        >>> d.add_function(lambda a, b: a + b, name="sum")
        <function __main__.<lambda>>

        Or use as decorator

        >>> d = Dispatcher()
        >>> @d.add_function
            def mymethod(*args, **kwargs):
                print(args, kwargs)

        Or use as a decorator with a different function name
        >>> d = Dispatcher()
        >>> @d.add_function(name="my.method")
            def mymethod(*args, **kwargs):
                print(args, kwargs)

        """
        if name and not f:
            return functools.partial(self.add_function, name=name)

        self[name or f.__name__] = f
        return f