| """ |
| 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}", |
| ]) |