Coverage for src/symphra_modules/loader/auto.py: 36.11%

1"""自动加载器实现.""" 

2 

3import os 

4from pathlib import Path 

5 

6from symphra_modules.abc import ModuleInterface 

7from symphra_modules.loader.directory import DirectoryLoader 

8from symphra_modules.loader.package import PackageLoader 

9from symphra_modules.utils import get_logger 

10 

11logger = get_logger() 

12 

13 

14class AutoLoader: 

15 """自动加载器，支持多种加载策略.""" 

16 

17 def __init__(self) -> None: 

18 """初始化自动加载器.""" 

19 self.loaders: dict[str, DirectoryLoader | PackageLoader] = { 

20 "directory": DirectoryLoader(), 

21 "package": PackageLoader(), 

22 } 

23 

24 def load_from_directory(self, directory: str | Path, recursive: bool = False) -> dict[str, type[ModuleInterface]]: 

25 """从目录加载模块. 

26 

27 Args: 

28 directory: 目录路径 

29 recursive: 是否递归加载子目录 

30 

31 Returns: 

32 模块类字典 

33 """ 

34 loader = DirectoryLoader(Path(directory)) 

35 

36 if recursive: 

37 all_modules: dict[str, type[ModuleInterface]] = {} 

38 for root, _dirs, _ in os.walk(directory): 

39 try: 

40 modules = loader.load(root) 

41 all_modules.update(modules) 

42 except Exception as e: 

43 logger.warning(f"无法从目录 {root} 加载模块: {e}") 

44 return all_modules 

45 

46 return loader.load(str(directory)) 

47 

48 def load_from_package(self, package_name: str) -> dict[str, type[ModuleInterface]]: 

49 """从包加载模块. 

50 

51 Args: 

52 package_name: 包名 

53 

54 Returns: 

55 模块类字典 

56 """ 

57 loader = self.loaders["package"] 

58 assert isinstance(loader, PackageLoader) 

59 return loader.load(package_name) 

60 

61 def discover_modules(self, source: str, source_type: str = "directory") -> list[str]: 

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

63 

64 Args: 

65 source: 模块源 

66 source_type: 源类型（"directory" 或 "package"） 

67 

68 Returns: 

69 模块名称列表 

70 

71 Raises: 

72 ValueError: 不支持的源类型 

73 """ 

74 if source_type not in self.loaders: 

75 raise ValueError(f"不支持的源类型: {source_type}") 

76 

77 return self.loaders[source_type].discover(source)