跳转至

事件系统

events

事件系统.

Classes

Event dataclass

事件基类.

Source code in src/symphra_modules/events/types.py 
10
11
12
13
14
15
16
17
@dataclass
class Event:
    """事件基类."""

    event_type: str  # 事件类型
    module_name: str  # 模块名称
    timestamp: datetime  # 时间戳
    data: dict[str, Any] | None = None  # 附加数据

EventBus

事件总线 - 简单的发布/订阅实现.

Source code in src/symphra_modules/events/bus.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class EventBus:
    """事件总线 - 简单的发布/订阅实现."""

    def __init__(self) -> None:
        """初始化事件总线."""
        self._handlers: dict[str, list[EventHandler]] = defaultdict(list)
        self._wildcard_handlers: list[EventHandler] = []

    def subscribe(self, event_type: str, handler: EventHandler) -> None:
        """订阅事件.

        Args:
            event_type: 事件类型，使用 "*" 订阅所有事件
            handler: 事件处理器函数

        Raises:
            EventError: 处理器不可调用时抛出
        """
        if not callable(handler):
            raise EventError("事件处理器必须是可调用对象")

        if event_type == "*":
            self._wildcard_handlers.append(handler)
        else:
            self._handlers[event_type].append(handler)

        logger.debug(f"订阅事件: {event_type}")

    def unsubscribe(self, event_type: str, handler: EventHandler) -> None:
        """取消订阅事件.

        Args:
            event_type: 事件类型
            handler: 事件处理器函数
        """
        if event_type == "*":
            if handler in self._wildcard_handlers:
                self._wildcard_handlers.remove(handler)
        else:
            if handler in self._handlers[event_type]:
                self._handlers[event_type].remove(handler)

        logger.debug(f"取消订阅事件: {event_type}")

    def publish(self, event: Event) -> None:
        """发布事件.

        Args:
            event: 事件对象
        """
        logger.debug(f"发布事件: {event.event_type} - {event.module_name}")

        # 调用通配符处理器
        for handler in self._wildcard_handlers:
            self._call_handler(handler, event)

        # 调用特定类型的处理器
        if event.event_type in self._handlers:
            for handler in self._handlers[event.event_type]:
                self._call_handler(handler, event)

    def _call_handler(self, handler: EventHandler, event: Event) -> None:
        """调用事件处理器.

        Args:
            handler: 事件处理器
            event: 事件对象
        """
        try:
            handler(event)
        except Exception as e:
            logger.error(f"事件处理器执行失败: {handler.__name__} - {e}")

    def clear(self) -> None:
        """清除所有订阅."""
        self._handlers.clear()
        self._wildcard_handlers.clear()
        logger.debug("已清除所有事件订阅")

    def get_subscribers(self, event_type: str) -> list[EventHandler]:
        """获取指定事件类型的订阅者.

        Args:
            event_type: 事件类型

        Returns:
            订阅者列表
        """
        if event_type == "*":
            return self._wildcard_handlers.copy()
        return self._handlers.get(event_type, []).copy()
Functions
__init__()

初始化事件总线.

源代码位于： src/symphra_modules/events/bus.py 
19
20
21
22
def __init__(self) -> None:
    """初始化事件总线."""
    self._handlers: dict[str, list[EventHandler]] = defaultdict(list)
    self._wildcard_handlers: list[EventHandler] = []
clear()

清除所有订阅.

源代码位于： src/symphra_modules/events/bus.py 
89
90
91
92
93
def clear(self) -> None:
    """清除所有订阅."""
    self._handlers.clear()
    self._wildcard_handlers.clear()
    logger.debug("已清除所有事件订阅")
get_subscribers(event_type)

获取指定事件类型的订阅者.

参数：

名称 类型 描述 默认
event_type str

事件类型

 必需

返回：

类型 描述
list[EventHandler]

订阅者列表
源代码位于： src/symphra_modules/events/bus.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def get_subscribers(self, event_type: str) -> list[EventHandler]:
    """获取指定事件类型的订阅者.

    Args:
        event_type: 事件类型

    Returns:
        订阅者列表
    """
    if event_type == "*":
        return self._wildcard_handlers.copy()
    return self._handlers.get(event_type, []).copy()
publish(event)

发布事件.

参数：

名称 类型 描述 默认
event Event

事件对象

 必需
源代码位于： src/symphra_modules/events/bus.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def publish(self, event: Event) -> None:
    """发布事件.

    Args:
        event: 事件对象
    """
    logger.debug(f"发布事件: {event.event_type} - {event.module_name}")

    # 调用通配符处理器
    for handler in self._wildcard_handlers:
        self._call_handler(handler, event)

    # 调用特定类型的处理器
    if event.event_type in self._handlers:
        for handler in self._handlers[event.event_type]:
            self._call_handler(handler, event)
subscribe(event_type, handler)

订阅事件.

参数：

名称 类型 描述 默认
event_type str

事件类型，使用 "*" 订阅所有事件

 必需
handler EventHandler

事件处理器函数

 必需

引发：

类型 描述
EventError

处理器不可调用时抛出
源代码位于： src/symphra_modules/events/bus.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def subscribe(self, event_type: str, handler: EventHandler) -> None:
    """订阅事件.

    Args:
        event_type: 事件类型，使用 "*" 订阅所有事件
        handler: 事件处理器函数

    Raises:
        EventError: 处理器不可调用时抛出
    """
    if not callable(handler):
        raise EventError("事件处理器必须是可调用对象")

    if event_type == "*":
        self._wildcard_handlers.append(handler)
    else:
        self._handlers[event_type].append(handler)

    logger.debug(f"订阅事件: {event_type}")
unsubscribe(event_type, handler)

取消订阅事件.

参数：

名称 类型 描述 默认
event_type str

事件类型

 必需
handler EventHandler

事件处理器函数

 必需
源代码位于： src/symphra_modules/events/bus.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def unsubscribe(self, event_type: str, handler: EventHandler) -> None:
    """取消订阅事件.

    Args:
        event_type: 事件类型
        handler: 事件处理器函数
    """
    if event_type == "*":
        if handler in self._wildcard_handlers:
            self._wildcard_handlers.remove(handler)
    else:
        if handler in self._handlers[event_type]:
            self._handlers[event_type].remove(handler)

    logger.debug(f"取消订阅事件: {event_type}")

ModuleErrorEvent dataclass

Bases: Event

模块错误事件.

Source code in src/symphra_modules/events/types.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
@dataclass
class ModuleErrorEvent(Event):
    """模块错误事件."""

    def __init__(
        self,
        module_name: str,
        error: Exception,
        data: dict[str, Any] | None = None,
    ) -> None:
        """初始化错误事件.

        Args:
            module_name: 模块名称
            error: 异常对象
            data: 附加数据（可选）
        """
        error_data = data or {}
        error_data["error"] = str(error)
        error_data["error_type"] = type(error).__name__

        super().__init__(
            event_type="module.error",
            module_name=module_name,
            timestamp=now(),
            data=error_data,
        )
Functions
__init__(module_name, error, data=None)

初始化错误事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
error Exception

异常对象

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def __init__(
    self,
    module_name: str,
    error: Exception,
    data: dict[str, Any] | None = None,
) -> None:
    """初始化错误事件.

    Args:
        module_name: 模块名称
        error: 异常对象
        data: 附加数据（可选）
    """
    error_data = data or {}
    error_data["error"] = str(error)
    error_data["error_type"] = type(error).__name__

    super().__init__(
        event_type="module.error",
        module_name=module_name,
        timestamp=now(),
        data=error_data,
    )

ModuleInstalledEvent dataclass

Bases: Event

模块安装事件.

Source code in src/symphra_modules/events/types.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
@dataclass
class ModuleInstalledEvent(Event):
    """模块安装事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化安装事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.installed",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化安装事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化安装事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.installed",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )

ModuleLoadedEvent dataclass

Bases: Event

模块加载事件.

Source code in src/symphra_modules/events/types.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
@dataclass
class ModuleLoadedEvent(Event):
    """模块加载事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化加载事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.loaded",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化加载事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化加载事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.loaded",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )

ModuleStartedEvent dataclass

Bases: Event

模块启动事件.

Source code in src/symphra_modules/events/types.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@dataclass
class ModuleStartedEvent(Event):
    """模块启动事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化启动事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.started",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化启动事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化启动事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.started",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )

ModuleStoppedEvent dataclass

Bases: Event

模块停止事件.

Source code in src/symphra_modules/events/types.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@dataclass
class ModuleStoppedEvent(Event):
    """模块停止事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化停止事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.stopped",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化停止事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化停止事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.stopped",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )

ModuleUninstalledEvent dataclass

Bases: Event

模块卸载事件.

Source code in src/symphra_modules/events/types.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
@dataclass
class ModuleUninstalledEvent(Event):
    """模块卸载事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化卸载事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.uninstalled",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化卸载事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化卸载事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.uninstalled",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )

ModuleUnregisteredEvent dataclass

Bases: Event

模块注销事件.

Source code in src/symphra_modules/events/types.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@dataclass
class ModuleUnregisteredEvent(Event):
    """模块注销事件."""

    def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
        """初始化注销事件.

        Args:
            module_name: 模块名称
            data: 附加数据（可选）
        """
        super().__init__(
            event_type="module.unregistered",
            module_name=module_name,
            timestamp=now(),
            data=data,
        )
Functions
__init__(module_name, data=None)

初始化注销事件.

参数：

名称 类型 描述 默认
module_name str

模块名称

 必需
data dict[str, Any] | None

附加数据（可选）

 None
源代码位于： src/symphra_modules/events/types.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
def __init__(self, module_name: str, data: dict[str, Any] | None = None) -> None:
    """初始化注销事件.

    Args:
        module_name: 模块名称
        data: 附加数据（可选）
    """
    super().__init__(
        event_type="module.unregistered",
        module_name=module_name,
        timestamp=now(),
        data=data,
    )