Coverage for src/symphra_modules/loader/base.py: 75.00%

32 statements  

« prev     ^ index     » next       coverage.py v7.11.0, created at 2025-10-26 18:16 +0800

1"""模块加载器基类.""" 

2 

3import inspect 

4from abc import ABC, abstractmethod 

5from typing import Any 

6 

7from symphra_modules.abc import BaseModule, ModuleInterface, ModuleMetadata 

8from symphra_modules.utils import get_logger 

9 

10logger = get_logger() 

11 

12 

13class ModuleLoader(ABC): 

14 """模块加载器抽象基类.""" 

15 

16 @abstractmethod 

17 def load(self, source: str) -> dict[str, type[ModuleInterface]]: 

18 """从源加载模块. 

19 

20 Args: 

21 source: 模块源(目录路径、包名等) 

22 

23 Returns: 

24 模块名到模块类的映射字典 

25 """ 

26 

27 @abstractmethod 

28 def discover(self, source: str) -> list[str]: 

29 """发现可用的模块. 

30 

31 Args: 

32 source: 模块源 

33 

34 Returns: 

35 模块名称列表 

36 """ 

37 

38 def _is_valid_module_class(self, obj: Any) -> bool: 

39 """检查对象是否是有效的模块类. 

40 

41 Args: 

42 obj: 待检查的对象 

43 

44 Returns: 

45 是否为有效的模块类 

46 """ 

47 # 跳过抽象类和自身 

48 if obj is BaseModule or inspect.isabstract(obj): 48 ↛ 49line 48 didn't jump to line 49 because the condition on line 48 was never true

49 return False 

50 

51 # 必须是类 

52 if not inspect.isclass(obj): 52 ↛ 53line 52 didn't jump to line 53 because the condition on line 52 was never true

53 return False 

54 

55 # 检查必需的属性和方法 

56 required_attrs = ["metadata", "install", "start", "stop"] 

57 if not all(hasattr(obj, attr) for attr in required_attrs): 57 ↛ 61line 57 didn't jump to line 61 because the condition on line 57 was always true

58 return False 

59 

60 # 检查 metadata 是否是 property 

61 return isinstance(getattr(type(obj), "metadata", None), property) 

62 

63 def _validate_module_instance(self, obj: type[ModuleInterface]) -> bool: 

64 """验证模块实例是否有效. 

65 

66 Args: 

67 obj: 模块类 

68 

69 Returns: 

70 是否可以成功实例化并获取元数据 

71 """ 

72 try: 

73 instance = obj() 

74 metadata = instance.metadata 

75 return isinstance(metadata, ModuleMetadata) 

76 except Exception as e: 

77 logger.warning(f"无法实例化模块类 {obj.__name__}: {e}") 

78 return False 

79 

80 def _find_module_classes(self, module: Any) -> dict[str, type[ModuleInterface]]: 

81 """在模块中查找有效的模块类. 

82 

83 Args: 

84 module: 要搜索的 Python 模块 

85 

86 Returns: 

87 模块名到模块类的映射字典 

88 """ 

89 module_classes: dict[str, type[ModuleInterface]] = {} 

90 

91 # 获取模块中的所有类 

92 for name, obj in inspect.getmembers(module, inspect.isclass): 

93 # 基本有效性检查 

94 if not self._is_valid_module_class(obj): 94 ↛ 98line 94 didn't jump to line 98 because the condition on line 94 was always true

95 continue 

96 

97 # 实例验证 

98 if self._validate_module_instance(obj): 

99 module_classes[name] = obj 

100 

101 return module_classes