autogen_core中的DataclassJsonMessageSerializer类
源代码
import json
from dataclasses import asdict, dataclass, fields
from typing import Any, ClassVar, Dict, List, Protocol, Sequence, TypeVar, cast, get_args, get_origin, runtime_checkable, Union
from pydantic import BaseModelfrom types import NoneType, UnionTypedef _type_name(cls: type[Any] | Any) -> str:if isinstance(cls, type):return cls.__name__else:return cast(str, cls.__class__.__name__)def is_union(t: object) -> bool:origin = get_origin(t)return origin is Union or origin is UnionTypeT = TypeVar("T")class MessageSerializer(Protocol[T]):@propertydef data_content_type(self) -> str: ...@propertydef type_name(self) -> str: ...def deserialize(self, payload: bytes) -> T: ...def serialize(self, message: T) -> bytes: ...@runtime_checkable
class IsDataclass(Protocol):# as already noted in comments, checking for this attribute is currently# the most reliable way to ascertain that something is a dataclass__dataclass_fields__: ClassVar[Dict[str, Any]]def is_dataclass(cls: type[Any]) -> bool:return hasattr(cls, "__dataclass_fields__")def has_nested_dataclass(cls: type[IsDataclass]) -> bool:# iterate fields and check if any of them are dataclassesreturn any(is_dataclass(f.type) for f in cls.__dataclass_fields__.values())def contains_a_union(cls: type[IsDataclass]) -> bool:return any(is_union(f.type) for f in cls.__dataclass_fields__.values())def has_nested_base_model(cls: type[IsDataclass]) -> bool:for f in fields(cls):field_type = f.type# Resolve forward references and other annotationsorigin = get_origin(field_type)args = get_args(field_type)# If the field type is directly a subclass of BaseModelif isinstance(field_type, type) and issubclass(field_type, BaseModel):return True# If the field type is a generic type like List[BaseModel], Tuple[BaseModel, ...], etc.if origin is not None and args:for arg in args:# Recursively check the argument typesif isinstance(arg, type) and issubclass(arg, BaseModel):return Trueelif get_origin(arg) is not None:# Handle nested generics like List[List[BaseModel]]if has_nested_base_model_in_type(arg):return True# Handle Union typeselif args:for arg in args:if isinstance(arg, type) and issubclass(arg, BaseModel):return Trueelif get_origin(arg) is not None:if has_nested_base_model_in_type(arg):return Truereturn Falsedef has_nested_base_model_in_type(tp: Any) -> bool:"""Helper function to check if a type or its arguments is a BaseModel subclass."""origin = get_origin(tp)args = get_args(tp)if isinstance(tp, type) and issubclass(tp, BaseModel):return Trueif origin is not None and args:for arg in args:if has_nested_base_model_in_type(arg):return Truereturn FalseDataclassT = TypeVar("DataclassT", bound=IsDataclass)JSON_DATA_CONTENT_TYPE = "application/json"
"""JSON data content type"""# TODO: what's the correct content type? There seems to be some disagreement over what it should be
PROTOBUF_DATA_CONTENT_TYPE = "application/x-protobuf"
"""Protobuf data content type"""class DataclassJsonMessageSerializer(MessageSerializer[DataclassT]):def __init__(self, cls: type[DataclassT]) -> None:if contains_a_union(cls):raise ValueError("Dataclass has a union type, which is not supported. To use a union, use a Pydantic model")if has_nested_dataclass(cls) or has_nested_base_model(cls):raise ValueError("Dataclass has nested dataclasses or base models, which are not supported. To use nested types, use a Pydantic model")self.cls = cls@propertydef data_content_type(self) -> str:return JSON_DATA_CONTENT_TYPE@propertydef type_name(self) -> str:return _type_name(self.cls)def deserialize(self, payload: bytes) -> DataclassT:message_str = payload.decode("utf-8")return self.cls(**json.loads(message_str))def serialize(self, message: DataclassT) -> bytes:return json.dumps(asdict(message)).encode("utf-8")
代码解释
这段代码定义了一个用于序列化和反序列化数据类(dataclass)和 Pydantic 模型的序列化器,特别是针对 JSON 格式。它的目标是提供一种结构化的方式来序列化和反序列化消息,并检查不支持的特性,例如嵌套的数据类、嵌套的 Pydantic 模型以及数据类中联合类型(union type)。
1. 导入
import json
from dataclasses import asdict, dataclass, fields
from typing import Any, ClassVar, Dict, List, Protocol, Sequence, TypeVar, cast, get_args, get_origin, runtime_checkable, Union
from pydantic import BaseModelfrom types import NoneType, UnionType
这部分导入了必要的模块:
json: 用于 JSON 序列化/反序列化。dataclasses: 用于处理数据类 (asdict,dataclass,fields)。typing: 用于类型提示和处理泛型 (Any,ClassVar,Dict,List,Protocol,Sequence,TypeVar,cast,get_args,get_origin,runtime_checkable,Union)。pydantic: 用于处理 Pydantic 模型 (BaseModel)。types: 用于处理类型提示,包括NoneType和UnionType。
2. 辅助函数
_type_name(cls): 返回类或对象的名称。is_union(t): 检查类型t是否为Union或UnionType。is_dataclass(cls): 检查类cls是否为数据类。has_nested_dataclass(cls): 检查数据类cls是否包含任何嵌套的数据类作为字段。contains_a_union(cls): 检查数据类cls是否包含任何Union类型作为字段类型。has_nested_base_model(cls): 检查数据类cls是否包含任何嵌套的 PydanticBaseModel子类作为字段,包括处理List[BaseModel]等泛型和嵌套泛型。has_nested_base_model_in_type(tp):has_nested_base_model使用的辅助函数,用于递归检查类型或其参数是否为BaseModel子类。
3. 类型变量和常量
T: 一个泛型类型变量。DataclassT: 绑定到IsDataclass的类型变量。JSON_DATA_CONTENT_TYPE: JSON 内容类型的常量。PROTOBUF_DATA_CONTENT_TYPE: Protobuf 内容类型的常量(虽然在 JSON 序列化器中未使用)。
4. MessageSerializer 协议
class MessageSerializer(Protocol[T]):@propertydef data_content_type(self) -> str: ...@propertydef type_name(self) -> str: ...def deserialize(self, payload: bytes) -> T: ...def serialize(self, message: T) -> bytes: ...
定义了消息序列化器的协议(接口)。任何实现此协议的类都必须定义指定的属性和方法:
data_content_type: 序列化数据的 MIME 类型。type_name: 被序列化类型的名称。deserialize(payload): 将字节负载反序列化为T类型的对象。serialize(message): 将T类型的对象序列化为字节负载。
5. IsDataclass 协议
@runtime_checkable
class IsDataclass(Protocol):__dataclass_fields__: ClassVar[Dict[str, Any]]
一个运行时可检查的协议,用于确定类是否为数据类。它检查 __dataclass_fields__ 属性是否存在。
6. DataclassJsonMessageSerializer 类
class DataclassJsonMessageSerializer(MessageSerializer[DataclassT]):# ...
此类实现了 MessageSerializer 协议,用于数据类的 JSON 序列化。
__init__(self, cls): 构造函数,它接受数据类类型cls作为参数。它执行关键检查:- 如果数据类包含
Union类型,则引发ValueError(使用 Pydantic 处理联合类型)。 - 如果数据类包含嵌套的数据类或 Pydantic 模型,则引发
ValueError(使用 Pydantic 处理嵌套类型)。
- 如果数据类包含
data_content_type属性:返回JSON_DATA_CONTENT_TYPE。type_name属性:返回数据类的名称。deserialize(payload): 将 JSON 负载反序列化为数据类的实例。serialize(message): 将数据类实例序列化为 JSON 负载。
总结
这段代码提供了一个专门用于简单数据类(没有联合或嵌套的数据类/模型)的 JSON 序列化器。它强调使用 Pydantic 来处理更复杂的场景,从而实现了清晰的关注点分离,并使序列化过程更具可预测性。这是一种很好的方法,可以避免直接在数据类中泛型序列化嵌套结构和联合类型的复杂性。
几个例子
@dataclass
class Point:x: inty: intserializer = DataclassJsonMessageSerializer(Point)point = Point(x=10, y=20)
serialized_data = serializer.serialize(point)
print(f"Serialized data: {serialized_data}") # Output: b'{"x": 10, "y": 20}'deserialized_point = serializer.deserialize(serialized_data)
print(f"Deserialized point: {deserialized_point}") # Output: Point(x=10, y=20)print(isinstance(deserialized_point, Point)) # Output: True
Serialized data: b'{"x": 10, "y": 20}'
Deserialized point: Point(x=10, y=20)
True
@dataclass
class Line:points: list[Point]line = Line(points=[Point(x=1, y=2), Point(x=3, y=4)])
serializer_line = DataclassJsonMessageSerializer(Line)serialized_line = serializer_line.serialize(line)
print(f"Serialized line: {serialized_line}") # Output: b'{"points": [{"x": 1, "y": 2}, {"x": 3, "y": 4}]}'deserialized_line = serializer_line.deserialize(serialized_line)
print(f"Deserialized line: {deserialized_line}") # Output: Line(points=[Point(x=1, y=2), Point(x=3, y=4)])print(isinstance(deserialized_line, Line)) # Output: True
Serialized line: b'{"points": [{"x": 1, "y": 2}, {"x": 3, "y": 4}]}'
Deserialized line: Line(points=[{'x': 1, 'y': 2}, {'x': 3, 'y': 4}])
True
@dataclass
class Rectangle:top_left: Pointbottom_right: Point# 这会抛出 ValueError,因为 Rectangle 包含嵌套的数据类 Point
try:serializer_rectangle = DataclassJsonMessageSerializer(Rectangle)
except ValueError as e:print(f"Error: {e}") # Output: Error: Dataclass has nested dataclasses or base models, which are not supported. To use nested types, use a Pydantic model
Error: Dataclass has nested dataclasses or base models, which are not supported. To use nested types, use a Pydantic model
from typing import Union@dataclass
class Shape:shape_type: Union[str, int]try:serializer_shape = DataclassJsonMessageSerializer(Shape)
except ValueError as e:print(f"Error: {e}") # Output: Error: Dataclass has a union type, which is not supported. To use a union, use a Pydantic model
Error: Dataclass has a union type, which is not supported. To use a union, use a Pydantic model
相关文章:
autogen_core中的DataclassJsonMessageSerializer类
源代码 import json from dataclasses import asdict, dataclass, fields from typing import Any, ClassVar, Dict, List, Protocol, Sequence, TypeVar, cast, get_args, get_origin, runtime_checkable, Union from pydantic import BaseModelfrom types import NoneType, …...
floodfill算法系列一>太平洋大西洋水流问题
目录 题目方法解析:代码设计:代码呈现: 题目方法解析: 代码设计: 代码呈现: class Solution {int m,n;int[] dx {0,0,-1,1};int[] dy {-1,1,0,0};public List<List<Integer>> pacificAtlant…...
DeepSeek 助力 Vue 开发:打造丝滑的无限滚动(Infinite Scroll)
前言:哈喽,大家好,今天给大家分享一篇文章!并提供具体代码帮助大家深入理解,彻底掌握!创作不易,如果能帮助到大家或者给大家一些灵感和启发,欢迎收藏关注哦 💕 目录 Deep…...
JavaScript 内置对象-Math对象
在JavaScript中,Math 对象提供了一系列与数学相关的静态方法和属性,帮助开发者执行复杂的计算任务。无论是简单的算术运算还是高级的几何、统计计算,Math 对象都能提供强大的支持。本文将详细介绍 Math 对象的主要功能及其使用方法。 一、简…...
硕成C语言22【一些算法和数组的概念】
1.求水仙花数 #include <stdio.h>int main() {//求水仙花数:1.三位数 2.个位的立方十位的立方百位的立方该数int unit, tens, hundreds;for (int i 100; i < 1000; i)//i表示该水仙花数{unit i / 1 % 10;tens i / 10 % 10;hundreds i / 100 % 10;if (…...
Halcon相机标定
1,前言。 相机的成像过程实质上是坐标系的转换。首先空间中的点由“世界坐标系”转换到“相机坐标系”,然后再将其投影到成像平面(图像物理坐标系),最后再将成像的平面上的数据转换为图像像素坐标系。但是由于透镜的制…...
部门管理(体验,最原始方法来做,Django+mysql)
本人初学,写完代码在此记录和复盘 在创建和注册完APP之后(我的命名是employees),编写models.py文件创建表 手动插入了几条数据 1.部门查询 urls.py和views.py在编写之前,都要注意导入对应的库 urls.py:…...
clickhouse集群搭建
Clickhouse集群搭建 文章目录 Clickhouse集群搭建安装包下载clickhouse单机安装默认安装默认数据库目录更改默认数据目录 2分片-1副本-3节点集群搭建1. 配置hosts2. 修改每个主机的主机名3. 配置文件上传配置文件分布chnode1配置文件chnode2配置文件chnode3配置文件 4. 重启cli…...
250214-java类集框架
单列集合是list和set,list的实现类有ArrayList和LinkedList,前者是数组实现,后者是链表实现。list和set,前者有序、可重复,后者无序不可重复。 1.单列集合 1.1. list java.util.List接口继承自Collection接口&#…...
二叉树(C语言版)
文章目录 二叉树完全二叉树和满二叉树二叉搜索树基本操作实现代码运行结果 分析红黑树2-3-4树(理论模型)红黑树(实际实现) 二叉树 树是一种层次结构,它在现实生活中是广泛存在的,比如:族谱(family tree),组织机构,目录…...
ASP.NET Core 面试宝典【刷题系列】
文章目录 引言1、什么是 dot net core 的 startup class?2、什么是中间件?3、application builder 的 use 和 run 方法有什么区别?4、dot net core 管道里面的map拓展有什么作用?5、dot net core 里面的路径是如何处理的?6、如何在 dot net core 中激活 session 功能?7、…...
案例-02.部门管理-查询
一.查询部门-需求 二.查询部门-思路 API接口文档 三.代码实现 1.controller层:负责与前端进行交互,接收前端所发来的请求 注:Slf4j用于记录日志使用,可以省略private static Logger log LoggerFactory.getLogger(DeptControlle…...
src和href区别
src和href区别 (1)请求资源类型不同(2)作用结果不同(3)解析方式不同 (1)请求资源类型不同 href 用来建立文档和元素之间的链接(是引用),常用的有a、linksrc 在请求src资源时候会将指向的资源下载并且应用到文档中(引入),常用的有script、iframe、image。 (2)作用结果不同 hr…...
Java每日精进·45天挑战·Day19
第一部分:移除数字以形成最小数的贪心算法实现 在编程的世界里,我们经常遇到需要对字符串表示的数字进行操作的问题。今天,我们要深入探讨一个具体的挑战:给定一个以字符串形式表示的非负整数 num 和一个整数 k,我们的…...
区块链的交易管理和共识机制
区块链的交易管理和共识机制是其核心功能,以下为你详细介绍它们的实现方式: 交易管理的实现 交易发起 • 用户使用钱包软件创建一笔交易,该交易包含发送方地址、接收方地址、转账金额等关键信息。同时,发送方会使用自己的私钥对…...
最新国内 ChatGPT Plus/Pro 获取教程
最后更新版本:20250202 教程介绍: 本文将详细介绍如何快速获取一张虚拟信用卡,并通过该卡来获取ChatGPT Plus和ChatGPT Pro。 # 教程全程约15分钟开通ChatGPT Plus会员帐号前准备工作 一个尚未升级的ChatGPT帐号!一张虚拟信用卡…...
Apollo 9.0 速度动态规划决策算法 – path time heuristic optimizer
文章目录 1. 动态规划2. 采样3. 代价函数3.1 障碍物代价3.2 距离终点代价3.3 速度代价3.4 加速度代价3.5 jerk代价 4. 回溯 这一章将来讲解速度决策算法,也就是SPEED_HEURISTIC_OPTIMIZER task里面的内容。Apollo 9.0使用动态规划算法进行速度决策,从类名…...
Apache Iceberg 与 Apache Hudi:数据湖领域的双雄对决
在数据存储和处理不断发展的领域中,数据湖仓的概念已经崭露头角,成为了一种变革性的力量。数据湖仓结合了数据仓库和数据湖的最佳元素,提供了一个统一的平台,支持数据科学、商业智能、人工智能/机器学习以及临时报告等多种关键功能…...
【LeetCode Hot100 普通数组】最大子数组和、合并区间、旋转数组、除自身以外数组的乘积、缺失的第一个正整数
普通数组 1. 最大子数组和(Maximum Subarray)解题思路动态规划的优化解法(Kadane 算法)核心思想 代码解析 2. 合并区间(Merge Intervals)解题思路代码实现 3. 旋转数组(Rotate Array)…...
共享存储-一步一步部署ceph分布式文件系统
一、Ceph 简介 Ceph 是一个开源的分布式文件系统。因为它还支持块存储、对象存储,所以很自 然的被用做云计算框架 openstack 或 cloudstack 整个存储后端。当然也可以单独作 为存储,例如部署一套集群作为对象存储、SAN 存储、NAS 存储等。 二、ceph 支…...
【WiFi帧结构】
文章目录 帧结构MAC头部管理帧 帧结构 Wi-Fi的帧分为三部分组成:MAC头部frame bodyFCS,其中MAC是固定格式的,frame body是可变长度。 MAC头部有frame control,duration,address1,address2,addre…...
剑指offer20_链表中环的入口节点
链表中环的入口节点 给定一个链表,若其中包含环,则输出环的入口节点。 若其中不包含环,则输出null。 数据范围 节点 val 值取值范围 [ 1 , 1000 ] [1,1000] [1,1000]。 节点 val 值各不相同。 链表长度 [ 0 , 500 ] [0,500] [0,500]。 …...
如何将联系人从 iPhone 转移到 Android
从 iPhone 换到 Android 手机时,你可能需要保留重要的数据,例如通讯录。好在,将通讯录从 iPhone 转移到 Android 手机非常简单,你可以从本文中学习 6 种可靠的方法,确保随时保持连接,不错过任何信息。 第 1…...
数据链路层的主要功能是什么
数据链路层(OSI模型第2层)的核心功能是在相邻网络节点(如交换机、主机)间提供可靠的数据帧传输服务,主要职责包括: 🔑 核心功能详解: 帧封装与解封装 封装: 将网络层下发…...
从零开始打造 OpenSTLinux 6.6 Yocto 系统(基于STM32CubeMX)(九)
设备树移植 和uboot设备树修改的内容同步到kernel将设备树stm32mp157d-stm32mp157daa1-mx.dts复制到内核源码目录下 源码修改及编译 修改arch/arm/boot/dts/st/Makefile,新增设备树编译 stm32mp157f-ev1-m4-examples.dtb \stm32mp157d-stm32mp157daa1-mx.dtb修改…...
新能源汽车智慧充电桩管理方案:新能源充电桩散热问题及消防安全监管方案
随着新能源汽车的快速普及,充电桩作为核心配套设施,其安全性与可靠性备受关注。然而,在高温、高负荷运行环境下,充电桩的散热问题与消防安全隐患日益凸显,成为制约行业发展的关键瓶颈。 如何通过智慧化管理手段优化散…...
【git】把本地更改提交远程新分支feature_g
创建并切换新分支 git checkout -b feature_g 添加并提交更改 git add . git commit -m “实现图片上传功能” 推送到远程 git push -u origin feature_g...
C++ Visual Studio 2017厂商给的源码没有.sln文件 易兆微芯片下载工具加开机动画下载。
1.先用Visual Studio 2017打开Yichip YC31xx loader.vcxproj,再用Visual Studio 2022打开。再保侟就有.sln文件了。 易兆微芯片下载工具加开机动画下载 ExtraDownloadFile1Info.\logo.bin|0|0|10D2000|0 MFC应用兼容CMD 在BOOL CYichipYC31xxloaderDlg::OnIni…...
【Oracle】分区表
个人主页:Guiat 归属专栏:Oracle 文章目录 1. 分区表基础概述1.1 分区表的概念与优势1.2 分区类型概览1.3 分区表的工作原理 2. 范围分区 (RANGE Partitioning)2.1 基础范围分区2.1.1 按日期范围分区2.1.2 按数值范围分区 2.2 间隔分区 (INTERVAL Partit…...
推荐 github 项目:GeminiImageApp(图片生成方向,可以做一定的素材)
推荐 github 项目:GeminiImageApp(图片生成方向,可以做一定的素材) 这个项目能干嘛? 使用 gemini 2.0 的 api 和 google 其他的 api 来做衍生处理 简化和优化了文生图和图生图的行为(我的最主要) 并且有一些目标检测和切割(我用不到) 视频和 imagefx 因为没 a…...
