150 lines
4.2 KiB
Python
150 lines
4.2 KiB
Python
"""
|
|
模块注册表模块
|
|
|
|
本模块提供统一的模块注册机制，用于管理所有业务模块和操作。
|
|
使用注册表模式可以：
|
|
- 统一管理所有操作
|
|
- 方便添加新功能模块
|
|
- 支持动态路由和操作查找
|
|
- 提供操作元数据管理
|
|
"""
|
|
from typing import Dict, Callable, Optional, Any
|
|
from dataclasses import dataclass
|
|
|
|
|
|
@dataclass
|
|
class ActionConfig:
|
|
"""
|
|
操作配置数据类
|
|
|
|
存储操作的配置信息，包括处理函数、所属模块、描述等。
|
|
|
|
属性:
|
|
handler: 处理函数，执行具体业务逻辑
|
|
module_name: 所属模块名称
|
|
description: 操作描述，用于文档和日志
|
|
requires_auth: 是否需要认证，默认 True
|
|
header_module: 使用的请求头模块名称，可选
|
|
"""
|
|
handler: Callable # 处理函数
|
|
module_name: str # 所属模块名称
|
|
description: Optional[str] = None # 描述
|
|
requires_auth: bool = True # 是否需要认证
|
|
header_module: Optional[str] = None # 使用的请求头模块名称
|
|
|
|
|
|
class ModuleRegistry:
|
|
"""
|
|
模块注册表类
|
|
|
|
统一管理所有业务模块和操作的注册表。
|
|
支持注册模块实例和操作，并提供查询功能。
|
|
|
|
属性:
|
|
_actions: 操作字典，格式为 {操作名: ActionConfig}
|
|
_modules: 模块字典，格式为 {模块名: 模块信息}
|
|
"""
|
|
|
|
def __init__(self):
|
|
"""初始化模块注册表"""
|
|
self._actions: Dict[str, ActionConfig] = {}
|
|
self._modules: Dict[str, Dict[str, Any]] = {}
|
|
|
|
def register_action(
|
|
self,
|
|
action_name: str,
|
|
handler: Callable,
|
|
module_name: str = "default",
|
|
description: Optional[str] = None,
|
|
requires_auth: bool = True,
|
|
header_module: Optional[str] = None
|
|
):
|
|
"""
|
|
注册操作
|
|
|
|
Args:
|
|
action_name: 操作名称
|
|
handler: 处理函数
|
|
module_name: 模块名称
|
|
description: 描述
|
|
requires_auth: 是否需要认证
|
|
header_module: 请求头模块名称
|
|
"""
|
|
config = ActionConfig(
|
|
handler=handler,
|
|
module_name=module_name,
|
|
description=description,
|
|
requires_auth=requires_auth,
|
|
header_module=header_module
|
|
)
|
|
self._actions[action_name] = config
|
|
|
|
def get_action(self, action_name: str) -> Optional[ActionConfig]:
|
|
"""
|
|
获取操作配置
|
|
|
|
Args:
|
|
action_name: 操作名称
|
|
|
|
Returns:
|
|
操作配置，如果不存在返回None
|
|
"""
|
|
return self._actions.get(action_name)
|
|
|
|
def get_all_actions(self) -> Dict[str, ActionConfig]:
|
|
"""
|
|
获取所有注册的操作
|
|
|
|
Returns:
|
|
Dict[str, ActionConfig]: 所有注册的操作字典的副本
|
|
"""
|
|
return self._actions.copy()
|
|
|
|
def register_module(self, module_name: str, module_instance: Any, **metadata):
|
|
"""
|
|
注册模块实例
|
|
|
|
Args:
|
|
module_name: 模块名称
|
|
module_instance: 模块实例
|
|
**metadata: 模块元数据
|
|
"""
|
|
self._modules[module_name] = {
|
|
'instance': module_instance,
|
|
**metadata
|
|
}
|
|
|
|
def get_module(self, module_name: str) -> Optional[Any]:
|
|
"""
|
|
获取模块实例
|
|
|
|
Args:
|
|
module_name: 模块名称
|
|
|
|
Returns:
|
|
模块实例，如果不存在返回None
|
|
"""
|
|
module_info = self._modules.get(module_name)
|
|
return module_info['instance'] if module_info else None
|
|
|
|
def get_actions_by_module(self, module_name: str) -> Dict[str, ActionConfig]:
|
|
"""
|
|
获取指定模块的所有操作
|
|
|
|
Args:
|
|
module_name: 模块名称
|
|
|
|
Returns:
|
|
操作字典
|
|
"""
|
|
return {
|
|
name: config
|
|
for name, config in self._actions.items()
|
|
if config.module_name == module_name
|
|
}
|
|
|
|
|
|
# 全局模块注册表实例
|
|
# 在应用启动时使用此实例来注册所有模块和操作
|
|
registry = ModuleRegistry()
|