1""" 2QMP Message Format 3 4This module provides the `Message` class, which represents a single QMP 5message sent to or from the server. 6""" 7 8import json 9from json import JSONDecodeError 10from typing import ( 11 Dict, 12 Iterator, 13 Mapping, 14 MutableMapping, 15 Optional, 16 Union, 17) 18 19from .error import ProtocolError 20 21 22class Message(MutableMapping[str, object]): 23 """ 24 Represents a single QMP protocol message. 25 26 QMP uses JSON objects as its basic communicative unit; so this 27 Python object is a :py:obj:`~collections.abc.MutableMapping`. It may 28 be instantiated from either another mapping (like a `dict`), or from 29 raw `bytes` that still need to be deserialized. 30 31 Once instantiated, it may be treated like any other MutableMapping:: 32 33 >>> msg = Message(b'{"hello": "world"}') 34 >>> assert msg['hello'] == 'world' 35 >>> msg['id'] = 'foobar' 36 >>> print(msg) 37 { 38 "hello": "world", 39 "id": "foobar" 40 } 41 42 It can be converted to `bytes`:: 43 44 >>> msg = Message({"hello": "world"}) 45 >>> print(bytes(msg)) 46 b'{"hello":"world","id":"foobar"}' 47 48 Or back into a garden-variety `dict`:: 49 50 >>> dict(msg) 51 {'hello': 'world'} 52 53 54 :param value: Initial value, if any. 55 :param eager: 56 When `True`, attempt to serialize or deserialize the initial value 57 immediately, so that conversion exceptions are raised during 58 the call to ``__init__()``. 59 """ 60 # pylint: disable=too-many-ancestors 61 62 def __init__(self, 63 value: Union[bytes, Mapping[str, object]] = b'{}', *, 64 eager: bool = True): 65 self._data: Optional[bytes] = None 66 self._obj: Optional[Dict[str, object]] = None 67 68 if isinstance(value, bytes): 69 self._data = value 70 if eager: 71 self._obj = self._deserialize(self._data) 72 else: 73 self._obj = dict(value) 74 if eager: 75 self._data = self._serialize(self._obj) 76 77 # Methods necessary to implement the MutableMapping interface, see: 78 # https://docs.python.org/3/library/collections.abc.html#collections.abc.MutableMapping 79 80 # We get pop, popitem, clear, update, setdefault, __contains__, 81 # keys, items, values, get, __eq__ and __ne__ for free. 82 83 def __getitem__(self, key: str) -> object: 84 return self._object[key] 85 86 def __setitem__(self, key: str, value: object) -> None: 87 self._object[key] = value 88 self._data = None 89 90 def __delitem__(self, key: str) -> None: 91 del self._object[key] 92 self._data = None 93 94 def __iter__(self) -> Iterator[str]: 95 return iter(self._object) 96 97 def __len__(self) -> int: 98 return len(self._object) 99 100 # Dunder methods not related to MutableMapping: 101 102 def __repr__(self) -> str: 103 if self._obj is not None: 104 return f"Message({self._object!r})" 105 return f"Message({bytes(self)!r})" 106 107 def __str__(self) -> str: 108 """Pretty-printed representation of this QMP message.""" 109 return json.dumps(self._object, indent=2) 110 111 def __bytes__(self) -> bytes: 112 """bytes representing this QMP message.""" 113 if self._data is None: 114 self._data = self._serialize(self._obj or {}) 115 return self._data 116 117 # Conversion Methods 118 119 @property 120 def _object(self) -> Dict[str, object]: 121 """ 122 A `dict` representing this QMP message. 123 124 Generated on-demand, if required. This property is private 125 because it returns an object that could be used to invalidate 126 the internal state of the `Message` object. 127 """ 128 if self._obj is None: 129 self._obj = self._deserialize(self._data or b'{}') 130 return self._obj 131 132 @classmethod 133 def _serialize(cls, value: object) -> bytes: 134 """ 135 Serialize a JSON object as `bytes`. 136 137 :raise ValueError: When the object cannot be serialized. 138 :raise TypeError: When the object cannot be serialized. 139 140 :return: `bytes` ready to be sent over the wire. 141 """ 142 return json.dumps(value, separators=(',', ':')).encode('utf-8') 143 144 @classmethod 145 def _deserialize(cls, data: bytes) -> Dict[str, object]: 146 """ 147 Deserialize JSON `bytes` into a native Python `dict`. 148 149 :raise DeserializationError: 150 If JSON deserialization fails for any reason. 151 :raise UnexpectedTypeError: 152 If the data does not represent a JSON object. 153 154 :return: A `dict` representing this QMP message. 155 """ 156 try: 157 obj = json.loads(data) 158 except JSONDecodeError as err: 159 emsg = "Failed to deserialize QMP message." 160 raise DeserializationError(emsg, data) from err 161 if not isinstance(obj, dict): 162 raise UnexpectedTypeError( 163 "QMP message is not a JSON object.", 164 obj 165 ) 166 return obj 167 168 169class DeserializationError(ProtocolError): 170 """ 171 A QMP message was not understood as JSON. 172 173 When this Exception is raised, ``__cause__`` will be set to the 174 `json.JSONDecodeError` Exception, which can be interrogated for 175 further details. 176 177 :param error_message: Human-readable string describing the error. 178 :param raw: The raw `bytes` that prompted the failure. 179 """ 180 def __init__(self, error_message: str, raw: bytes): 181 super().__init__(error_message) 182 #: The raw `bytes` that were not understood as JSON. 183 self.raw: bytes = raw 184 185 def __str__(self) -> str: 186 return "\n".join([ 187 super().__str__(), 188 f" raw bytes were: {str(self.raw)}", 189 ]) 190 191 192class UnexpectedTypeError(ProtocolError): 193 """ 194 A QMP message was JSON, but not a JSON object. 195 196 :param error_message: Human-readable string describing the error. 197 :param value: The deserialized JSON value that wasn't an object. 198 """ 199 def __init__(self, error_message: str, value: object): 200 super().__init__(error_message) 201 #: The JSON value that was expected to be an object. 202 self.value: object = value 203 204 def __str__(self) -> str: 205 strval = json.dumps(self.value, indent=2) 206 return "\n".join([ 207 super().__str__(), 208 f" json value was: {strval}", 209 ]) 210