diff options
Diffstat (limited to 'python/qemu/aqmp/message.py')
| -rw-r--r-- | python/qemu/aqmp/message.py | 209 |
1 files changed, 0 insertions, 209 deletions
diff --git a/python/qemu/aqmp/message.py b/python/qemu/aqmp/message.py deleted file mode 100644 index f76ccc9074..0000000000 --- a/python/qemu/aqmp/message.py +++ /dev/null @@ -1,209 +0,0 @@ -""" -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}", - ]) |