autogen_core._topic 源代码

import re
from dataclasses import dataclass

from typing_extensions import Self


def is_valid_topic_type(value: str) -> bool:
    return bool(re.match(r"^[\w\-\.\:\=]+\Z", value))




@dataclass(eq=True, frozen=True)
class TopicId:
    """
    TopicId 定义了广播消息的作用范围。本质上，代理运行时通过其广播 API 实现了发布-订阅模型：发布消息时必须指定主题。

    更多信息请参阅：:ref:`topic_and_subscription_topic`
    """

    type: str
    """此 topic_id 包含的事件类型。遵循云事件规范。

    必须匹配模式：^[\\w\\-\\.\\:\\=]+\\Z

    了解更多：https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#type
    """

    source: str
    """标识事件发生的上下文环境。遵循云事件规范。

    了解更多：https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md#source-1
    """

    def __post_init__(self) -> None:
        if is_valid_topic_type(self.type) is False:
            raise ValueError(f"Invalid topic type: {self.type}. Must match the pattern: ^[\\w\\-\\.\\:\\=]+\\Z")

    def __str__(self) -> str:
        return f"{self.type}/{self.source}"



    @classmethod
    def from_str(cls, topic_id: str) -> Self:
        """将格式为 ``type/source`` 的字符串转换为 TopicId"""
        items = topic_id.split("/", maxsplit=1)
        if len(items) != 2:
            raise ValueError(f"Invalid topic id: {topic_id}")
        type, source = items[0], items[1]
        return cls(type, source)