10 - 载荷转换

载荷转换器（Payload Converter）负责将应用程序对象序列化为载荷（Payload），及将其反序列化。载荷是带有元数据的二进制形式，Temporal 用其传输数据。

默认情况下，Temporal 使用 DefaultPayloadConverter，它可以处理 None、bytes、Protobuf 消息以及任何可 JSON 序列化的数据。仅当应用程序使用原生不支持的类型时，才需要自定义载荷转换器。

默认支持的类型

默认的数据转换器支持转换多种类型，包括：

None

bytes

google.protobuf.message.Message —— 编码时以 JSON 形式处理，但可以解码来自其他语言的二进制 proto

任何可以转换为 JSON 的数据，包括：
- json.dump 原生支持的任何类型
- dataclasses
- 可迭代对象，包括 JSON dump 默认不支持的类型，比如 set
- 基于 IntEnum、StrEnum 的枚举类型
- UUID

尽管工作流、更新、信号和查询都可以定义多个输入参数，但强烈建议使用单个 dataclass 或 Pydantic 模型参数，以便在后续添加带默认值的字段时不破坏兼容性。类似的建议也适用于返回值。

使用泛型的类可能无法正确解析其泛型类型。当前实现不支持泛型类型解析。应使用具体类型。

使用 Pydantic 模型

要使用 Pydantic 模型实例，请先安装 Pydantic，然后在创建 Client 实例时设置 Pydantic 数据转换器：

from temporalio.contrib.pydantic import pydantic_data_converter

client = Client(data_converter=pydantic_data_converter, ...)

该数据转换器支持将 Pydantic 支持的所有类型与 JSON 之间进行相互转换。除 Pydantic 模型外，支持的类型还包括：

json.dumps() 默认支持的所有类型

json.dumps() 不支持但属于标准库的若干类型，包括 dataclasses、datetime 模块中的类型、set、UUID 等

由上述任意类型组合而成的自定义类型，且支持任意层级的嵌套。比如，包含 datetime 字段的 Pydantic 模型列表。

datetime.date、datetime.time 和 datetime.datetime 仅能与 Pydantic 数据转换器一起使用。

默认转换器的工作方式

默认转换器是 CompositePayloadConverter，它按顺序尝试每种编码转换器，直到某个转换器能够处理该值。在序列化时，按顺序依次尝试每个 EncodingPayloadConverter，直到其中一个成功为止。

载荷转换器可以独立于载荷编解码器（Payload Codec）进行自定义。

自定义载荷转换器

要处理自定义数据类型，需要创建新 EncodingPayloadConverter。比如，要支持 IPv4Address 类型：

class IPv4AddressEncodingPayloadConverter(EncodingPayloadConverter):
    @property
    def encoding(self) -> str:
        return "text/ipv4-address"

    def to_payload(self, value: Any) -> Optional[Payload]:
        if isinstance(value, ipaddress.IPv4Address):
            return Payload(
                metadata={"encoding": self.encoding.encode()},
                data=str(value).encode(),
            )
        else:
            return None

    def from_payload(self, payload: Payload, type_hint: Optional[Type] = None) -> Any:
        assert not type_hint or type_hint is ipaddress.IPv4Address
        return ipaddress.IPv4Address(payload.data.decode())

class IPv4AddressPayloadConverter(CompositePayloadConverter):
    def __init__(self) -> None:
# Just add ours as first before the defaults
        super().__init__(
            IPv4AddressEncodingPayloadConverter(),
            *DefaultPayloadConverter.default_encoding_payload_converters,
        )

my_data_converter = dataclasses.replace(
    DataConverter.default,
    payload_converter_className=IPv4AddressPayloadConverter,
)

为自定义类型定制 JSON 转换器

如果希望自定义类型能够用于列表、联合及其他集合中，那么应该对现有的 JSON 转换器进行定制，而非新增编码转换器。JSON 转换器位于转换器列表的最后，因此它负责处理其他转换器无法识别的类型。

使用自定义的 json.JSONEncoder 定制序列化，使用自定义的 JSONTypeConverter 定制反序列化。

class IPv4AddressJSONEncoder(AdvancedJSONEncoder):
    def default(self, o: Any) -> Any:
        if isinstance(o, ipaddress.IPv4Address):
            return str(o)
        return super().default(o)

class IPv4AddressJSONTypeConverter(JSONTypeConverter):
    def to_typed_value(
        self, hint: Type, value: Any
    ) -> Union[Optional[Any], _JSONTypeConverterUnhandled]:
        if issubclass(hint, ipaddress.IPv4Address):
            return ipaddress.IPv4Address(value)
        return JSONTypeConverter.Unhandled

class IPv4AddressPayloadConverter(CompositePayloadConverter):
    def __init__(self) -> None:
# Replace default JSON plain with our own that has our encoder and type
# converter
        json_converter = JSONPlainPayloadConverter(
            encoder=IPv4AddressJSONEncoder,
            custom_type_converters=[IPv4AddressJSONTypeConverter()],
        )
        super().__init__(
            *[
                c if not isinstance(c, JSONPlainPayloadConverter) else json_converter
                for c in DefaultPayloadConverter.default_encoding_payload_converters
            ]
        )

my_data_converter = dataclasses.replace(
    DataConverter.default,
    payload_converter_className=IPv4AddressPayloadConverter,
)

现在，可以在类型提示中使用 IPv4Address，包括集合、可选类型等。