From 2f68ecac7186521e0aef28aabf5d00dfe9081eb2 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Fri, 12 Sep 2025 16:50:43 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20massage=5Fitem=20=E5=AE=8C=E5=96=84?= =?UTF-8?q?=20content=20=E5=AD=97=E6=AE=B5=20from=5Fdict=20=E8=BD=AC?= =?UTF-8?q?=E6=8D=A2=E6=93=8D=E4=BD=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/queue_sqlite/model/message_item.py | 197 +++++++++++++------------ 1 file changed, 105 insertions(+), 92 deletions(-) diff --git a/src/queue_sqlite/model/message_item.py b/src/queue_sqlite/model/message_item.py index 0cfae71..4f44dcd 100644 --- a/src/queue_sqlite/model/message_item.py +++ b/src/queue_sqlite/model/message_item.py @@ -13,70 +13,70 @@ class MessageItem: content: dict = field( metadata={"description": "消息内容，包含具体的任务数据或信息"} ) - + # 可选字段 status: MessageStatus = field( default=MessageStatus.PENDING, - metadata={"description": "消息状态：PENDING(待处理)、PROCESSING(处理中)、COMPLETED(已完成)、FAILED(失败)"} + metadata={ + "description": "消息状态：PENDING(待处理)、PROCESSING(处理中)、COMPLETED(已完成)、FAILED(失败)" + }, ) type: MessageType = field( default=MessageType.TASK, - metadata={"description": "消息类型：TASK(任务)、NOTIFICATION(通知)、RESPONSE(响应)"} + metadata={ + "description": "消息类型：TASK(任务)、NOTIFICATION(通知)、RESPONSE(响应)" + }, ) createtime: datetime = field( - default_factory=datetime.now, # type: ignore - metadata={"description": "消息创建时间"} + default_factory=datetime.now, # type: ignore + metadata={"description": "消息创建时间"}, ) updatetime: datetime = field( - default_factory=datetime.now, # type: ignore - metadata={"description": "消息最后更新时间"} + default_factory=datetime.now, # type: ignore + metadata={"description": "消息最后更新时间"}, ) result: dict = field( default_factory=dict, - metadata={"description": "消息处理结果，存储任务执行后的输出数据"} + metadata={"description": "消息处理结果，存储任务执行后的输出数据"}, ) id: str = field( default_factory=lambda: str(uuid.uuid4()), - metadata={"description": "消息唯一标识符"} + metadata={"description": "消息唯一标识符"}, ) priority: MessagePriority = field( default=MessagePriority.NORMAL, - metadata={"description": "消息优先级：LOW(低)、NORMAL(普通)、HIGH(高)、URGENT(紧急)"} + metadata={ + "description": "消息优先级：LOW(低)、NORMAL(普通)、HIGH(高)、URGENT(紧急)" + }, ) source: str = field( - default="client", - metadata={"description": "消息来源，标识发送方"} + default="client", metadata={"description": "消息来源，标识发送方"} ) destination: str = field( - default="client", - metadata={"description": "消息目标，标识接收方(即task函数名)"} + default="client", metadata={"description": "消息目标，标识接收方(即task函数名)"} ) retry_count: int = field( - default=0, - metadata={"description": "重试次数，记录消息处理失败后的重试次数"} + default=0, metadata={"description": "重试次数，记录消息处理失败后的重试次数"} ) expire_time: Optional[datetime] = field( - default=None, - metadata={"description": "消息过期时间，超过此时间消息将不再处理"} + default=None, metadata={"description": "消息过期时间，超过此时间消息将不再处理"} ) tags: Optional[str] = field( - default=None, - metadata={"description": "消息标签，用于分类和筛选"} + default=None, metadata={"description": "消息标签，用于分类和筛选"} ) metadata: dict = field( default_factory=dict, - metadata={"description": "额外元数据，存储自定义的扩展信息"} + metadata={"description": "额外元数据，存储自定义的扩展信息"}, ) - @classmethod def from_dict(cls, data: Dict[str, Any]) -> "MessageItem": """从字典创建消息对象""" # 处理日期时间字段 - datetime_fields = ['createtime', 'updatetime', 'expire_time'] + datetime_fields = ["createtime", "updatetime", "expire_time"] for field in datetime_fields: if field in data: - if data[field] == 'null': + if data[field] == "null": data[field] = None continue if isinstance(data[field], str): @@ -86,90 +86,103 @@ class MessageItem: elif data[field] is None: data[field] = None else: - raise ValueError(f"请使用 'str' 或 'datetime' 类型设置 {field} 字段") - - if 'result' in data: - if data['result'] == 'null': - data['result'] = {} - elif isinstance(data['result'], str): - data['result'] = json.loads(data['result']) + raise ValueError( + f"请使用 'str' 或 'datetime' 类型设置 {field} 字段" + ) - if 'tags' in data: - if data['tags'] == 'null': - data['tags'] = None + if "content" in data: + if data["content"] == "null": + data["content"] = {} + elif isinstance(data["content"], str): + data["content"] = json.loads(data["content"]) + + if "metadata" in data: + if data["metadata"] == "null": + data["metadata"] = {} + elif isinstance(data["metadata"], str): + data["metadata"] = json.loads(data["metadata"]) + + if "result" in data: + if data["result"] == "null": + data["result"] = {} + elif isinstance(data["result"], str): + data["result"] = json.loads(data["result"]) + + if "tags" in data: + if data["tags"] == "null": + data["tags"] = None # 处理枚举字段 - if 'type' in data: - if isinstance(data['type'], str): - data['type'] = MessageType(data['type']) # 使用枚举名称 - elif isinstance(data['type'], MessageType): + if "type" in data: + if isinstance(data["type"], str): + data["type"] = MessageType(data["type"]) # 使用枚举名称 + elif isinstance(data["type"], MessageType): pass # 已经是枚举类型 else: raise ValueError(f"请使用 'str' 或 'MessageType' 类型设置 type 字段") - - if 'status' in data: - if isinstance(data['status'], str): - data['status'] = MessageStatus(data['status']) # 使用枚举名称 - elif isinstance(data['status'], int): - data['status'] = MessageStatus(data['status']) # 使用枚举值 - elif isinstance(data['status'], MessageStatus): + + if "status" in data: + if isinstance(data["status"], str): + data["status"] = MessageStatus(data["status"]) # 使用枚举名称 + elif isinstance(data["status"], int): + data["status"] = MessageStatus(data["status"]) # 使用枚举值 + elif isinstance(data["status"], MessageStatus): pass # 已经是枚举类型 else: raise ValueError(f"Invalid message status: {data['status']}") - - if 'priority' in data: - if isinstance(data['priority'], (int, str)): - data['priority'] = MessagePriority(data['priority']) # type: ignore - elif isinstance(data['priority'], MessagePriority): + + if "priority" in data: + if isinstance(data["priority"], (int, str)): + data["priority"] = MessagePriority(data["priority"]) # type: ignore + elif isinstance(data["priority"], MessagePriority): pass else: raise ValueError("请使用 'int' 或 'str' 类型设置 priority 字段") - + return cls(**data) - + def to_dict(self) -> Dict[str, Any]: """转换为字典 - + Returns: Dict[str, Any]: 消息数据字典 """ return { - 'id': self.id, - 'type': self.type.value, - 'status': self.status.value, - 'content': self.content, - 'createtime': self.createtime.isoformat(), - 'updatetime': self.updatetime.isoformat(), - 'priority': self.priority.value, - 'source': self.source, - 'retry_count': self.retry_count, - 'expire_time': self.expire_time.isoformat() if self.expire_time else None, - 'tags': self.tags, - 'metadata': self.metadata, - 'result': self.result, - "destination": self.destination + "id": self.id, + "type": self.type.value, + "status": self.status.value, + "content": self.content, + "createtime": self.createtime.isoformat(), + "updatetime": self.updatetime.isoformat(), + "priority": self.priority.value, + "source": self.source, + "retry_count": self.retry_count, + "expire_time": self.expire_time.isoformat() if self.expire_time else None, + "tags": self.tags, + "metadata": self.metadata, + "result": self.result, + "destination": self.destination, } def to_dict_by_core(self) -> Dict[str, Any]: """转换为核心库需要的字典格式""" return { - 'id': self.id, - 'type': self.type.value, - 'status': self.status.value, - 'content': json.dumps(self.content), - 'createtime': self.createtime.isoformat(), - 'updatetime': self.updatetime.isoformat(), - 'priority': self.priority.value, - 'source': self.source, - 'retry_count': self.retry_count, - 'expire_time': self.expire_time.isoformat() if self.expire_time else 'null', - 'tags': self.tags if self.tags else 'null', - 'metadata': json.dumps(self.metadata), - 'result': json.dumps(self.result), - "destination": self.destination + "id": self.id, + "type": self.type.value, + "status": self.status.value, + "content": json.dumps(self.content), + "createtime": self.createtime.isoformat(), + "updatetime": self.updatetime.isoformat(), + "priority": self.priority.value, + "source": self.source, + "retry_count": self.retry_count, + "expire_time": self.expire_time.isoformat() if self.expire_time else "null", + "tags": self.tags if self.tags else "null", + "metadata": json.dumps(self.metadata), + "result": json.dumps(self.result), + "destination": self.destination, } - def to_json(self) -> str: """转换为json字符串""" # 将枚举类型转换为字符串 @@ -183,17 +196,17 @@ class MessageItem: def is_expired(self) -> bool: """检查消息是否过期 - + Returns: bool: 是否过期 """ if self.expire_time is None: return False return datetime.now() > self.expire_time - + def can_retry(self) -> bool: """检查消息是否可以重试 - + Returns: bool: 是否可以重试 """ @@ -202,26 +215,26 @@ class MessageItem: @classmethod def get_field_descriptions(cls) -> Dict[str, str]: """获取所有字段的描述信息 - + Returns: Dict[str, str]: 字段名到描述信息的映射 """ from dataclasses import fields - + descriptions = {} for field_info in fields(cls): - if 'description' in field_info.metadata: - descriptions[field_info.name] = field_info.metadata['description'] + if "description" in field_info.metadata: + descriptions[field_info.name] = field_info.metadata["description"] else: descriptions[field_info.name] = "无描述信息" - + return descriptions @classmethod def print_field_info(cls): """打印所有字段的信息，包括类型、默认值和描述""" from dataclasses import fields - + print(f"=== {cls.__name__} 字段信息 ===") for field_info in fields(cls): print(f"字段名: {field_info.name}") @@ -229,6 +242,6 @@ class MessageItem: print(f"默认值: {field_info.default}") if field_info.default_factory is not None: print(f"默认工厂: {field_info.default_factory}") - if 'description' in field_info.metadata: + if "description" in field_info.metadata: print(f"描述: {field_info.metadata['description']}") print("-" * 50)