GroupedMessage

类私有源码

gpm = syncstream.base.GroupedMessage(
    data: Sequence[str] | Warning | Exception | None = None,
)

一组信息。

用来封装多行str（信息）、单个的警告对象、或单个的异常对象。

该类自动用于mproc.LineProcMirror和host.LineHostMirror。绝大多数情况下，用户不需要直接使用该类。

参数

输入

参数	类型	必选	说明
data	Sequence[str] | Warning | Exception		一组信息来源。该输入值会处理成一行或多行。处理后的信息可以序列化。

方法

serialize

jdata: dict[str, Any] = gpm.serialize()

将该信息项序列化为JSON兼容的字典。

输出

参数	类型	说明
jdata	dict[str, Any]	序列化后的字典。该字典可以由deserialize方法逆变换为GroupedMessage。

---

deserialize

gpm: GroupedMessage = GroupedMessage.deserialize(jdata: dict[str, Any])

将JSON兼容的字典逆序列化为该类的实例。

输入

参数	类型	必选	说明
jdata	dict[str, Any]		序列化的字典。该字典JSON兼容，并且可以透过网络数据包传输。

输出

参数	类型	说明
gpm	GroupedMessage	逆序列化后的对象。该对象可以透过str(gpm)格式化为字符串。

范例

将错误信息的回溯捕获为一组信息

代码

use_grouped_messages.py

from syncstream.base import GroupedMessage

try:
   raise ImportError('A testing exception object.')
except ImportError as error:
   gpm = GroupedMessage(error)

print(gpm)

结果

Traceback (most recent call last):
  File "C:\Documents (no sync)\GitHub\sync-stream\alpha\test.py", line 4, in <module>
    raise ImportError('A testing exception object.')
ImportError: A testing exception object.