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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
|
"""
QMP Message Format
This module provides the `Message` class, which represents a single QMP
message sent to or from the server.
"""
import json
from json import JSONDecodeError
from typing import (
Dict,
Iterator,
Mapping,
MutableMapping,
Optional,
Union,
)
from .error import ProtocolError
class Message(MutableMapping[str, object]):
"""
Represents a single QMP protocol message.
QMP uses JSON objects as its basic communicative unit; so this
Python object is a :py:obj:`~collections.abc.MutableMapping`. It may
be instantiated from either another mapping (like a `dict`), or from
raw `bytes` that still need to be deserialized.
Once instantiated, it may be treated like any other MutableMapping::
>>> msg = Message(b'{"hello": "world"}')
>>> assert msg['hello'] == 'world'
>>> msg['id'] = 'foobar'
>>> print(msg)
{
"hello": "world",
"id": "foobar"
}
It can be converted to `bytes`::
>>> msg = Message({"hello": "world"})
>>> print(bytes(msg))
b'{"hello":"world","id":"foobar"}'
Or back into a garden-variety `dict`::
>>> dict(msg)
{'hello': 'world'}
:param value: Initial value, if any.
:param eager:
When `True`, attempt to serialize or deserialize the initial value
immediately, so that conversion exceptions are raised during
the call to ``__init__()``.
"""
# pylint: disable=too-many-ancestors
def __init__(self,
value: Union[bytes, Mapping[str, object]] = b'{}', *,
eager: bool = True):
self._data: Optional[bytes] = None
self._obj: Optional[Dict[str, object]] = None
if isinstance(value, bytes):
self._data = value
if eager:
self._obj = self._deserialize(self._data)
else:
self._obj = dict(value)
if eager:
self._data = self._serialize(self._obj)
# Methods necessary to implement the MutableMapping interface, see:
# https://docs.python.org/3/library/collections.abc.html#collections.abc.MutableMapping
# We get pop, popitem, clear, update, setdefault, __contains__,
# keys, items, values, get, __eq__ and __ne__ for free.
def __getitem__(self, key: str) -> object:
return self._object[key]
def __setitem__(self, key: str, value: object) -> None:
self._object[key] = value
self._data = None
def __delitem__(self, key: str) -> None:
del self._object[key]
self._data = None
def __iter__(self) -> Iterator[str]:
return iter(self._object)
def __len__(self) -> int:
return len(self._object)
# Dunder methods not related to MutableMapping:
def __repr__(self) -> str:
if self._obj is not None:
return f"Message({self._object!r})"
return f"Message({bytes(self)!r})"
def __str__(self) -> str:
"""Pretty-printed representation of this QMP message."""
return json.dumps(self._object, indent=2)
def __bytes__(self) -> bytes:
"""bytes representing this QMP message."""
if self._data is None:
self._data = self._serialize(self._obj or {})
return self._data
# Conversion Methods
@property
def _object(self) -> Dict[str, object]:
"""
A `dict` representing this QMP message.
Generated on-demand, if required. This property is private
because it returns an object that could be used to invalidate
the internal state of the `Message` object.
"""
if self._obj is None:
self._obj = self._deserialize(self._data or b'{}')
return self._obj
@classmethod
def _serialize(cls, value: object) -> bytes:
"""
Serialize a JSON object as `bytes`.
:raise ValueError: When the object cannot be serialized.
:raise TypeError: When the object cannot be serialized.
:return: `bytes` ready to be sent over the wire.
"""
return json.dumps(value, separators=(',', ':')).encode('utf-8')
@classmethod
def _deserialize(cls, data: bytes) -> Dict[str, object]:
"""
Deserialize JSON `bytes` into a native Python `dict`.
:raise DeserializationError:
If JSON deserialization fails for any reason.
:raise UnexpectedTypeError:
If the data does not represent a JSON object.
:return: A `dict` representing this QMP message.
"""
try:
obj = json.loads(data)
except JSONDecodeError as err:
emsg = "Failed to deserialize QMP message."
raise DeserializationError(emsg, data) from err
if not isinstance(obj, dict):
raise UnexpectedTypeError(
"QMP message is not a JSON object.",
obj
)
return obj
class DeserializationError(ProtocolError):
"""
A QMP message was not understood as JSON.
When this Exception is raised, ``__cause__`` will be set to the
`json.JSONDecodeError` Exception, which can be interrogated for
further details.
:param error_message: Human-readable string describing the error.
:param raw: The raw `bytes` that prompted the failure.
"""
def __init__(self, error_message: str, raw: bytes):
super().__init__(error_message)
#: The raw `bytes` that were not understood as JSON.
self.raw: bytes = raw
def __str__(self) -> str:
return "\n".join([
super().__str__(),
f" raw bytes were: {str(self.raw)}",
])
class UnexpectedTypeError(ProtocolError):
"""
A QMP message was JSON, but not a JSON object.
:param error_message: Human-readable string describing the error.
:param value: The deserialized JSON value that wasn't an object.
"""
def __init__(self, error_message: str, value: object):
super().__init__(error_message)
#: The JSON value that was expected to be an object.
self.value: object = value
def __str__(self) -> str:
strval = json.dumps(self.value, indent=2)
return "\n".join([
super().__str__(),
f" json value was: {strval}",
])
|
