File: base.py

package info (click to toggle)
python-jsonrpc 1.12.1-1
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 444 kB
  • sloc: python: 3,152; makefile: 154
file content (87 lines) | stat: -rw-r--r-- 2,034 bytes parent folder | download | duplicates (4)
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
from .utils import JSONSerializable


class JSONRPCBaseRequest(JSONSerializable):

    """ Base class for JSON-RPC 1.0 and JSON-RPC 2.0 requests."""

    def __init__(self, method=None, params=None, _id=None,
                 is_notification=None):
        self.data = dict()
        self.method = method
        self.params = params
        self._id = _id
        self.is_notification = is_notification

    @property
    def data(self):
        return self._data

    @data.setter
    def data(self, value):
        if not isinstance(value, dict):
            raise ValueError("data should be dict")

        self._data = value

    @property
    def args(self):
        """ Method position arguments.

        :return tuple args: method position arguments.

        """
        return tuple(self.params) if isinstance(self.params, list) else ()

    @property
    def kwargs(self):
        """ Method named arguments.

        :return dict kwargs: method named arguments.

        """
        return self.params if isinstance(self.params, dict) else {}

    @property
    def json(self):
        return self.serialize(self.data)


class JSONRPCBaseResponse(JSONSerializable):

    """ Base class for JSON-RPC 1.0 and JSON-RPC 2.0 responses."""

    def __init__(self, **kwargs):
        self.data = dict()

        try:
            self.result = kwargs['result']
        except KeyError:
            pass

        try:
            self.error = kwargs['error']
        except KeyError:
            pass

        self._id = kwargs.get('_id')

        if 'result' not in kwargs and 'error' not in kwargs:
            raise ValueError("Either result or error should be used")

        self.request = None  # type: JSONRPCBaseRequest

    @property
    def data(self):
        return self._data

    @data.setter
    def data(self, value):
        if not isinstance(value, dict):
            raise ValueError("data should be dict")

        self._data = value

    @property
    def json(self):
        return self.serialize(self.data)