Coverage for src/symphra_modules/registry.py: 82.96%

1"""模块注册表实现.""" 

2 

3from typing import Any 

4 

5from symphra_modules.abc import BaseModule, ModuleInterface 

6from symphra_modules.config import ModuleInfo, ModuleState 

7from symphra_modules.exceptions import ModuleNotFoundException, ModuleStateError 

8from symphra_modules.utils import get_logger, now 

9 

10logger = get_logger() 

11 

12 

13class ModuleRegistry: 

14 """模块注册表，管理模块实例和生命周期.""" 

15 

16 def __init__(self) -> None: 

17 """初始化注册表.""" 

18 self._modules: dict[str, ModuleInterface] = {} 

19 self._module_info: dict[str, ModuleInfo] = {} 

20 

21 def register( 

22 self, name: str, module_class: type[BaseModule], config: dict[str, Any] | None = None 

23 ) -> ModuleInterface: 

24 """注册模块. 

25 

26 Args: 

27 name: 模块名称 

28 module_class: 模块类 

29 config: 初始配置（可选） 

30 

31 Returns: 

32 模块实例 

33 

34 Raises: 

35 ModuleStateError: 模块已注册时抛出 

36 """ 

37 if name in self._modules: 

38 raise ModuleStateError(f"模块 {name} 已经注册") 

39 

40 module_instance = module_class(config) 

41 module_instance.bootstrap() # 调用引导方法 

42 

43 # 创建模块信息 

44 module_info = ModuleInfo( 

45 metadata=module_instance.metadata, 

46 state=ModuleState.LOADED, 

47 loaded_at=now(), 

48 ) 

49 self._modules[name] = module_instance 

50 self._module_info[name] = module_info 

51 logger.info(f"模块 {name} 注册成功") 

52 

53 return module_instance 

54 

55 def unregister(self, name: str) -> None: 

56 """注销模块. 

57 

58 Args: 

59 name: 模块名称 

60 """ 

61 _, info = self._get_module_and_info(name) 

62 if info.state == ModuleState.STARTED: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 self.stop(name) 

64 

65 del self._modules[name] 

66 del self._module_info[name] 

67 

68 logger.info(f"模块 {name} 注销成功") 

69 

70 def install(self, name: str, config: dict[str, Any] | None = None) -> None: 

71 """安装模块. 

72 

73 Args: 

74 name: 模块名称 

75 config: 安装配置（可选） 

76 

77 Raises: 

78 ModuleStateError: 状态不正确时抛出 

79 """ 

80 module, info = self._get_module_and_info(name) 

81 

82 # 已加载的模块才允许安装 

83 if info.state != ModuleState.LOADED: 

84 raise ModuleStateError( 

85 f"模块 {name} 状态错误，当前状态为 {info.state}，期望状态为 {ModuleState.LOADED}", 

86 module_name=name, 

87 current_state=info.state, 

88 expected_states=[ModuleState.LOADED], 

89 ) 

90 

91 try: 

92 module.install(config) 

93 info.state = ModuleState.INSTALLED 

94 info.installed_at = now() 

95 

96 if config: 

97 module.configure(config) 

98 info.config = config 

99 

100 logger.info(f"模块 {name} 安装成功") 

101 except Exception as e: 

102 info.state = ModuleState.ERROR 

103 raise ModuleStateError(f"模块 {name} 安装失败: {e}", module_name=name) from e 

104 

105 def uninstall(self, name: str) -> None: 

106 """卸载模块. 

107 

108 Args: 

109 name: 模块名称 

110 

111 Raises: 

112 ModuleStateError: 状态不正确时抛出 

113 """ 

114 module, info = self._get_module_and_info(name) 

115 

116 if info.state not in [ 116 ↛ 120line 116 didn't jump to line 120 because the condition on line 116 was never true

117 ModuleState.INSTALLED, 

118 ModuleState.STOPPED, 

119 ]: 

120 raise ModuleStateError( 

121 f"无法卸载模块 {name}，当前状态为 {info.state}", 

122 module_name=name, 

123 current_state=info.state, 

124 expected_states=[ModuleState.INSTALLED, ModuleState.STOPPED], 

125 ) 

126 try: 

127 module.uninstall() 

128 info.state = ModuleState.LOADED 

129 info.installed_at = None 

130 info.started_at = None 

131 logger.info(f"模块 {name} 卸载成功") 

132 except Exception as e: 

133 info.state = ModuleState.ERROR 

134 raise ModuleStateError(f"模块 {name} 卸载失败: {e}", module_name=name) from e 

135 

136 def start(self, name: str) -> None: 

137 """启动模块. 

138 

139 Args: 

140 name: 模块名称 

141 

142 Raises: 

143 ModuleStateError: 状态不正确时抛出 

144 """ 

145 module, info = self._get_module_and_info(name) 

146 

147 if info.state not in [ModuleState.INSTALLED, ModuleState.STOPPED]: 147 ↛ 148line 147 didn't jump to line 148 because the condition on line 147 was never true

148 raise ModuleStateError( 

149 f"无法启动模块 {name}，当前状态为 {info.state}", 

150 module_name=name, 

151 current_state=info.state, 

152 expected_states=[ModuleState.INSTALLED, ModuleState.STOPPED], 

153 ) 

154 try: 

155 module.start() 

156 info.state = ModuleState.STARTED 

157 info.started_at = now() 

158 

159 logger.info(f"模块 {name} 启动成功") 

160 

161 except Exception as e: 

162 info.state = ModuleState.ERROR 

163 raise ModuleStateError(f"模块 {name} 启动失败: {e}", module_name=name) from e 

164 

165 def stop(self, name: str) -> None: 

166 """停止模块. 

167 

168 Args: 

169 name: 模块名称 

170 

171 Raises: 

172 ModuleStateError: 停止失败时抛出 

173 """ 

174 module, info = self._get_module_and_info(name) 

175 

176 if info.state != ModuleState.STARTED: 

177 return # 模块未启动，无需停止 

178 

179 try: 

180 module.stop() 

181 info.state = ModuleState.STOPPED 

182 info.started_at = None 

183 

184 logger.info(f"模块 {name} 停止成功") 

185 

186 except Exception as e: 

187 info.state = ModuleState.ERROR 

188 raise ModuleStateError(f"模块 {name} 停止失败: {e}", module_name=name) from e 

189 

190 def reload(self, name: str) -> None: 

191 """重载模块. 

192 

193 Args: 

194 name: 模块名称 

195 

196 Raises: 

197 ModuleStateError: 重载失败时抛出 

198 """ 

199 module, _ = self._get_module_and_info(name) 

200 try: 

201 module.reload() 

202 logger.info(f"模块 {name} 重载成功") 

203 except Exception as e: 

204 self._module_info[name].state = ModuleState.ERROR 

205 raise ModuleStateError(f"模块 {name} 重载失败: {e}", module_name=name) from e 

206 

207 def configure(self, name: str, config: dict[str, Any] | None = None) -> None: 

208 """配置模块. 

209 

210 Args: 

211 name: 模块名称 

212 config: 配置字典（可选） 

213 

214 Raises: 

215 ModuleStateError: 配置失败时抛出 

216 """ 

217 module, info = self._get_module_and_info(name) 

218 

219 try: 

220 module.configure(config) 

221 info.config = config or {} 

222 logger.info(f"模块 {name} 配置成功") 

223 except Exception as e: 

224 raise ModuleStateError(f"模块 {name} 配置失败: {e}", module_name=name) from e 

225 

226 def get(self, name: str) -> ModuleInterface | None: 

227 """获取模块实例. 

228 

229 Args: 

230 name: 模块名称 

231 

232 Returns: 

233 模块实例，如果不存在则返回 None 

234 """ 

235 return self._modules.get(name) 

236 

237 def get_info(self, name: str) -> ModuleInfo | None: 

238 """获取模块信息. 

239 

240 Args: 

241 name: 模块名称 

242 

243 Returns: 

244 模块信息，如果不存在则返回 None 

245 """ 

246 return self._module_info.get(name) 

247 

248 def list_modules(self) -> list[str]: 

249 """列出所有已加载模块的名称. 

250 

251 Returns: 

252 模块名称列表 

253 """ 

254 return list(self._modules.keys()) 

255 

256 def list_modules_by_state(self, state: ModuleState) -> list[str]: 

257 """根据状态列出模块名称. 

258 

259 Args: 

260 state: 模块状态 

261 

262 Returns: 

263 符合状态的模块名称列表 

264 """ 

265 return [name for name, info in self._module_info.items() if info.state == state] 

266 

267 def get_module_states(self) -> dict[str, ModuleState]: 

268 """获取所有模块状态. 

269 

270 Returns: 

271 模块名到状态的映射 

272 """ 

273 return {name: info.state for name, info in self._module_info.items()} 

274 

275 def is_installed(self, name: str) -> bool: 

276 """检查模块是否已安装. 

277 

278 Args: 

279 name: 模块名称 

280 

281 Returns: 

282 是否已安装 

283 """ 

284 info = self.get_info(name) 

285 return info is not None and info.state in [ 

286 ModuleState.INSTALLED, 

287 ModuleState.STARTED, 

288 ModuleState.STOPPED, 

289 ] 

290 

291 def is_loaded(self, name: str) -> bool: 

292 """检查模块是否已加载. 

293 

294 Args: 

295 name: 模块名称 

296 

297 Returns: 

298 是否已加载 

299 """ 

300 return name in self._modules 

301 

302 def is_started(self, name: str) -> bool: 

303 """检查模块是否已启动. 

304 

305 Args: 

306 name: 模块名称 

307 

308 Returns: 

309 是否已启动 

310 """ 

311 info = self.get_info(name) 

312 return info is not None and info.state == ModuleState.STARTED 

313 

314 def _ensure_module_loaded(self, name: str) -> None: 

315 """确保模块已加载. 

316 

317 Args: 

318 name: 模块名称 

319 

320 Raises: 

321 ModuleNotFoundException: 模块未注册时抛出 

322 """ 

323 if name not in self._modules: 

324 raise ModuleNotFoundException(f"模块 {name} 未注册", module_name=name) 

325 

326 def _get_module_and_info(self, name: str) -> tuple[ModuleInterface, ModuleInfo]: 

327 """获取模块实例和信息. 

328 

329 Args: 

330 name: 模块名称 

331 

332 Returns: 

333 (模块实例, 模块信息)元组 

334 

335 Raises: 

336 ModuleNotFoundException: 模块未注册时抛出 

337 """ 

338 self._ensure_module_loaded(name) 

339 return self._modules[name], self._module_info[name]