跳转至

核心抽象 (ABC)

abc

模块接口和基类定义.

Classes

BaseModule

Bases: ABC

模块基类，提供默认实现.

Source code in src/symphra_modules/abc.py 
 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
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
class BaseModule(ABC):
    """模块基类，提供默认实现."""

    def __init__(self, config: dict[str, Any] | None = None) -> None:
        """初始化模块.

        Args:
            config: 初始配置（可选）
        """
        self._config = config or {}

    # 使用 __slots__ 降低每个模块实例的内存占用
    __slots__ = ("_config",)

    @property
    @abstractmethod
    def metadata(self) -> ModuleMetadata:
        """获取模块元数据 - 子类必须实现."""

    def bootstrap(self) -> None:
        """模块引导，注册依赖等操作.

        默认实现为空，子类可覆盖此方法执行初始化逻辑。
        """
        pass

    def install(self, config: dict[str, Any] | None = None) -> None:
        """安装模块的默认实现.

        Args:
            config: 安装配置（可选）
        """
        if config:
            self.configure(config)

    def uninstall(self) -> None:
        """卸载模块的默认实现.

        默认行为是先停止模块。
        """
        self.stop()

    def configure(self, config: dict[str, Any] | None = None) -> None:
        """配置模块.

        Args:
            config: 配置字典（可选）

        Raises:
            ModuleConfigError: 配置验证失败时抛出
        """
        if config and self.validate_config(config):
            self._config.update(config)
        elif config:
            from symphra_modules.exceptions import ModuleConfigError

            raise ModuleConfigError(
                f"配置验证失败: {self.metadata.name}",
                module_name=self.metadata.name,
            )

    def start(self) -> None:
        """启动模块的默认实现.

        子类应覆盖此方法实现实际的启动逻辑。
        """
        pass

    def stop(self) -> None:
        """停止模块的默认实现.

        子类应覆盖此方法实现实际的停止逻辑。
        """
        pass

    def reload(self) -> None:
        """重载模块的默认实现.

        默认行为是先停止再启动。
        """
        self.stop()
        self.start()

    def get_config(self) -> dict[str, Any]:
        """获取当前配置.

        Returns:
            配置字典的副本
        """
        return self._config.copy() if self._config else {}

    def validate_config(self, config: dict[str, Any] | None = None) -> bool:
        """验证配置 - 默认实现.

        默认总是返回 True，子类应覆盖此方法实现实际的验证逻辑。

        Args:
            config: 待验证的配置（可选）

        Returns:
            验证是否通过
        """
        return True
Attributes
metadata abstractmethod property

获取模块元数据 - 子类必须实现.

Functions
__init__(config=None)

初始化模块.

参数：

名称 类型 描述 默认
config dict[str, Any] | None

初始配置（可选）

 None
源代码位于： src/symphra_modules/abc.py 
60
61
62
63
64
65
66
def __init__(self, config: dict[str, Any] | None = None) -> None:
    """初始化模块.

    Args:
        config: 初始配置（可选）
    """
    self._config = config or {}
bootstrap()

模块引导，注册依赖等操作.

默认实现为空，子类可覆盖此方法执行初始化逻辑。

源代码位于： src/symphra_modules/abc.py 
76
77
78
79
80
81
def bootstrap(self) -> None:
    """模块引导，注册依赖等操作.

    默认实现为空，子类可覆盖此方法执行初始化逻辑。
    """
    pass
configure(config=None)

配置模块.

参数：

名称 类型 描述 默认
config dict[str, Any] | None

配置字典（可选）

 None

引发：

类型 描述
ModuleConfigError

配置验证失败时抛出
源代码位于： src/symphra_modules/abc.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def configure(self, config: dict[str, Any] | None = None) -> None:
    """配置模块.

    Args:
        config: 配置字典（可选）

    Raises:
        ModuleConfigError: 配置验证失败时抛出
    """
    if config and self.validate_config(config):
        self._config.update(config)
    elif config:
        from symphra_modules.exceptions import ModuleConfigError

        raise ModuleConfigError(
            f"配置验证失败: {self.metadata.name}",
            module_name=self.metadata.name,
        )
get_config()

获取当前配置.

返回：

类型 描述
dict[str, Any]

配置字典的副本
源代码位于： src/symphra_modules/abc.py 
140
141
142
143
144
145
146
def get_config(self) -> dict[str, Any]:
    """获取当前配置.

    Returns:
        配置字典的副本
    """
    return self._config.copy() if self._config else {}
install(config=None)

安装模块的默认实现.

参数：

名称 类型 描述 默认
config dict[str, Any] | None

安装配置（可选）

 None
源代码位于： src/symphra_modules/abc.py 
83
84
85
86
87
88
89
90
def install(self, config: dict[str, Any] | None = None) -> None:
    """安装模块的默认实现.

    Args:
        config: 安装配置（可选）
    """
    if config:
        self.configure(config)
reload()

重载模块的默认实现.

默认行为是先停止再启动。

源代码位于： src/symphra_modules/abc.py 
132
133
134
135
136
137
138
def reload(self) -> None:
    """重载模块的默认实现.

    默认行为是先停止再启动。
    """
    self.stop()
    self.start()
start()

启动模块的默认实现.

子类应覆盖此方法实现实际的启动逻辑。

源代码位于： src/symphra_modules/abc.py 
118
119
120
121
122
123
def start(self) -> None:
    """启动模块的默认实现.

    子类应覆盖此方法实现实际的启动逻辑。
    """
    pass
stop()

停止模块的默认实现.

子类应覆盖此方法实现实际的停止逻辑。

源代码位于： src/symphra_modules/abc.py 
125
126
127
128
129
130
def stop(self) -> None:
    """停止模块的默认实现.

    子类应覆盖此方法实现实际的停止逻辑。
    """
    pass
uninstall()

卸载模块的默认实现.

默认行为是先停止模块。

源代码位于： src/symphra_modules/abc.py 
92
93
94
95
96
97
def uninstall(self) -> None:
    """卸载模块的默认实现.

    默认行为是先停止模块。
    """
    self.stop()
validate_config(config=None)

验证配置 - 默认实现.

默认总是返回 True，子类应覆盖此方法实现实际的验证逻辑。

参数：

名称 类型 描述 默认
config dict[str, Any] | None

待验证的配置（可选）

 None

返回：

类型 描述
bool

验证是否通过
源代码位于： src/symphra_modules/abc.py 
148
149
150
151
152
153
154
155
156
157
158
159
def validate_config(self, config: dict[str, Any] | None = None) -> bool:
    """验证配置 - 默认实现.

    默认总是返回 True，子类应覆盖此方法实现实际的验证逻辑。

    Args:
        config: 待验证的配置（可选）

    Returns:
        验证是否通过
    """
    return True

ModuleInterface

Bases: Protocol

模块接口协议.

Source code in src/symphra_modules/abc.py 
11
12
13
14
15
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
@runtime_checkable
class ModuleInterface(Protocol):
    """模块接口协议."""

    @property
    def metadata(self) -> ModuleMetadata:
        """获取模块元数据."""
        ...

    def bootstrap(self) -> None:
        """模块引导，注册依赖等操作."""
        ...

    def install(self, config: dict[str, Any] | None = None) -> None:
        """安装模块."""
        ...

    def uninstall(self) -> None:
        """卸载模块."""
        ...

    def configure(self, config: dict[str, Any] | None = None) -> None:
        """配置模块."""
        ...

    def start(self) -> None:
        """启动模块."""
        ...

    def stop(self) -> None:
        """停止模块."""
        ...

    def reload(self) -> None:
        """重载模块."""
        ...

    def get_config(self) -> dict[str, Any]:
        """获取当前配置."""
        ...

    def validate_config(self, config: dict[str, Any] | None = None) -> bool:
        """验证配置."""
        ...
Attributes
metadata property

获取模块元数据.

Functions
bootstrap()

模块引导，注册依赖等操作.

源代码位于： src/symphra_modules/abc.py 
20
21
22
def bootstrap(self) -> None:
    """模块引导，注册依赖等操作."""
    ...
configure(config=None)

配置模块.

源代码位于： src/symphra_modules/abc.py 
32
33
34
def configure(self, config: dict[str, Any] | None = None) -> None:
    """配置模块."""
    ...
get_config()

获取当前配置.

源代码位于： src/symphra_modules/abc.py 
48
49
50
def get_config(self) -> dict[str, Any]:
    """获取当前配置."""
    ...
install(config=None)

安装模块.

源代码位于： src/symphra_modules/abc.py 
24
25
26
def install(self, config: dict[str, Any] | None = None) -> None:
    """安装模块."""
    ...
reload()

重载模块.

源代码位于： src/symphra_modules/abc.py 
44
45
46
def reload(self) -> None:
    """重载模块."""
    ...
start()

启动模块.

源代码位于： src/symphra_modules/abc.py 
36
37
38
def start(self) -> None:
    """启动模块."""
    ...
stop()

停止模块.

源代码位于： src/symphra_modules/abc.py 
40
41
42
def stop(self) -> None:
    """停止模块."""
    ...
uninstall()

卸载模块.

源代码位于： src/symphra_modules/abc.py 
28
29
30
def uninstall(self) -> None:
    """卸载模块."""
    ...
validate_config(config=None)

验证配置.

源代码位于： src/symphra_modules/abc.py 
52
53
54
def validate_config(self, config: dict[str, Any] | None = None) -> bool:
    """验证配置."""
    ...

Functions

call_module_method(module, method_name, *args, **kwargs) async

统一调用模块方法（自动检测同步/异步）.

参数：

名称 类型 描述 默认
module object

模块实例

 必需
method_name str

方法名称

 必需
*args Any

位置参数

 ()
**kwargs Any

关键字参数

 {}

返回：

类型 描述
Any

方法返回值

引发：

类型 描述
AttributeError

方法不存在时抛出
源代码位于： src/symphra_modules/abc.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
async def call_module_method(
    module: object,
    method_name: str,
    *args: Any,
    **kwargs: Any,
) -> Any:
    """统一调用模块方法（自动检测同步/异步）.

    Args:
        module: 模块实例
        method_name: 方法名称
        *args: 位置参数
        **kwargs: 关键字参数

    Returns:
        方法返回值

    Raises:
        AttributeError: 方法不存在时抛出
    """
    if not hasattr(module, method_name):
        raise AttributeError(f"模块 {type(module).__name__} 没有方法 {method_name}")

    method = getattr(module, method_name)

    # 如果是协程函数，直接 await
    if inspect.iscoroutinefunction(method):
        return await method(*args, **kwargs)

    # 如果是同步函数，在执行器中运行
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, lambda: method(*args, **kwargs))

is_async_module(module)

检测模块是否实现了异步方法.

参数：

名称 类型 描述 默认
module object

要检测的模块实例

 必需

返回：

类型 描述
bool

如果模块的核心方法是协程函数则返回 True
源代码位于： src/symphra_modules/abc.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def is_async_module(module: object) -> bool:
    """检测模块是否实现了异步方法.

    Args:
        module: 要检测的模块实例

    Returns:
        如果模块的核心方法是协程函数则返回 True
    """
    # 检查关键方法是否为协程函数
    key_methods = ["install", "uninstall", "start", "stop", "configure"]
    for method_name in key_methods:
        if hasattr(module, method_name):
            method = getattr(module, method_name)
            if inspect.iscoroutinefunction(method):
                return True
    return False