跳转至

加载器

loader

模块加载器.

Classes

DirectoryLoader

Bases: ModuleLoader

从目录加载模块.

Source code in src/symphra_modules/loader/directory.py 
 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
 55
 56
 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
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
class DirectoryLoader(ModuleLoader):
    """从目录加载模块."""

    def __init__(self, base_path: Path | None = None) -> None:
        """初始化目录加载器.

        Args:
            base_path: 基础路径，默认为当前工作目录
        """
        self.base_path = base_path or Path.cwd()

    def _to_module_name(self, path: Path) -> str | None:
        """将文件或包路径转换为模块名（相对 base_path）.

        Args:
            path: 文件或包路径

        Returns:
            模块名，如果无法转换则返回 None
        """
        try:
            relative = path.relative_to(self.base_path)
        except ValueError:
            # 无法转换为相对路径时返回 None
            return None

        parts = list(relative.parts)
        if parts and parts[-1].endswith(".py"):
            parts[-1] = parts[-1][:-3]

        return ".".join(parts)

    def load(self, source: str) -> dict[str, type[ModuleInterface]]:
        """从指定目录加载所有模块.

        Args:
            source: 目录路径（相对于 base_path）

        Returns:
            模块名到模块类的映射字典

        Raises:
            ModuleLoadError: 目录不存在时抛出
        """
        modules: dict[str, type[ModuleInterface]] = {}
        module_dir = Path(self.base_path) / source

        if not module_dir.exists():
            raise ModuleLoadError(f"模块目录不存在: {module_dir}")

        # 查找所有 Python 文件
        for py_file in module_dir.glob("*.py"):
            if py_file.name.startswith("_"):
                continue

            try:
                module_classes = self._load_from_file(py_file)
                modules.update(module_classes)
            except Exception as e:
                logger.warning(f"无法从 {py_file} 加载模块: {e}")

        # 查找包
        for package_dir in module_dir.iterdir():
            if package_dir.is_dir() and (package_dir / "__init__.py").exists():
                try:
                    module_classes = self._load_from_package(package_dir)
                    modules.update(module_classes)
                except Exception as e:
                    logger.error(f"无法从 {package_dir} 加载模块: {e}")

        return modules

    def discover(self, source: str) -> list[str]:
        """发现目录中的模块.

        Args:
            source: 目录路径

        Returns:
            模块名称列表
        """
        module_names: list[str] = []
        module_dir = Path(self.base_path) / source

        if not module_dir.exists():
            return module_names

        # Python 文件
        for py_file in module_dir.glob("*.py"):
            if not py_file.name.startswith("_"):
                module_names.append(py_file.stem)

        # 包
        for package_dir in module_dir.iterdir():
            if package_dir.is_dir() and (package_dir / "__init__.py").exists():
                module_names.append(package_dir.name)

        return module_names

    def _load_from_file(self, file_path: Path, package: str | None = None) -> dict[str, type[ModuleInterface]]:
        """从 Python 文件加载模块.

        Args:
            file_path: Python 文件路径
            package: 包名（可选）

        Returns:
            模块类字典

        Raises:
            ModuleLoadError: 加载失败时抛出
        """
        module_name: str
        if package is not None:
            module_name = f"{package}.{file_path.stem}"
        else:
            inferred = self._to_module_name(file_path)
            module_name = inferred or file_path.stem

        module: Any = None
        if "." in module_name:
            try:
                module = importlib.import_module(module_name)
            except ImportError:
                module = None

        if module is None:
            spec = importlib.util.spec_from_file_location(module_name, file_path)
            if not spec or not spec.loader:
                raise ModuleLoadError(f"无法从 {file_path} 创建模块规范")

            module = importlib.util.module_from_spec(spec)
            sys.modules[module_name] = module
            spec.loader.exec_module(module)

        return self._find_module_classes(module)

    def _load_from_package(self, package_dir: Path) -> dict[str, type[ModuleInterface]]:
        """从包加载模块.

        Args:
            package_dir: 包目录路径

        Returns:
            模块类字典

        Raises:
            ModuleLoadError: 加载失败时抛出
        """
        inferred_package = self._to_module_name(package_dir)
        package_name = inferred_package or package_dir.name

        # 将包路径添加到 sys.path，便于定位依赖
        parent_path = str(package_dir.parent)
        if parent_path not in sys.path:
            sys.path.insert(0, parent_path)

        modules: dict[str, type[ModuleInterface]] = {}

        # 优先加载显式的 module.py，再加载其他非私有 Python 文件
        candidate_files: list[Path] = []
        module_file = package_dir / "module.py"
        if module_file.exists():
            candidate_files.append(module_file)

        if not candidate_files:
            # 回退：若没有独立文件，则尝试加载包本身
            try:
                module = importlib.import_module(package_name)
                return self._find_module_classes(module)
            except ImportError as e:
                raise ModuleLoadError(f"无法从 {package_name} 加载模块: {e}") from e

        for py_file in candidate_files:
            try:
                module_classes = self._load_from_file(py_file, package=package_name)
                modules.update(module_classes)
            except Exception as e:
                logger.warning(f"无法从 {package_name} 的 {py_file.name} 加载模块: {e}")

        return modules
Functions
__init__(base_path=None)

初始化目录加载器.

参数：

名称 类型 描述 默认
base_path Path | None

基础路径，默认为当前工作目录

 None
源代码位于： src/symphra_modules/loader/directory.py 
20
21
22
23
24
25
26
def __init__(self, base_path: Path | None = None) -> None:
    """初始化目录加载器.

    Args:
        base_path: 基础路径，默认为当前工作目录
    """
    self.base_path = base_path or Path.cwd()
discover(source)

发现目录中的模块.

参数：

名称 类型 描述 默认
source str

目录路径

 必需

返回：

类型 描述
list[str]

模块名称列表
源代码位于： src/symphra_modules/loader/directory.py 
 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
def discover(self, source: str) -> list[str]:
    """发现目录中的模块.

    Args:
        source: 目录路径

    Returns:
        模块名称列表
    """
    module_names: list[str] = []
    module_dir = Path(self.base_path) / source

    if not module_dir.exists():
        return module_names

    # Python 文件
    for py_file in module_dir.glob("*.py"):
        if not py_file.name.startswith("_"):
            module_names.append(py_file.stem)

    # 包
    for package_dir in module_dir.iterdir():
        if package_dir.is_dir() and (package_dir / "__init__.py").exists():
            module_names.append(package_dir.name)

    return module_names
load(source)

从指定目录加载所有模块.

参数：

名称 类型 描述 默认
source str

目录路径（相对于 base_path）

 必需

返回：

类型 描述
dict[str, type[ModuleInterface]]

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

引发：

类型 描述
ModuleLoadError

目录不存在时抛出
源代码位于： src/symphra_modules/loader/directory.py 
49
50
51
52
53
54
55
56
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
def load(self, source: str) -> dict[str, type[ModuleInterface]]:
    """从指定目录加载所有模块.

    Args:
        source: 目录路径（相对于 base_path）

    Returns:
        模块名到模块类的映射字典

    Raises:
        ModuleLoadError: 目录不存在时抛出
    """
    modules: dict[str, type[ModuleInterface]] = {}
    module_dir = Path(self.base_path) / source

    if not module_dir.exists():
        raise ModuleLoadError(f"模块目录不存在: {module_dir}")

    # 查找所有 Python 文件
    for py_file in module_dir.glob("*.py"):
        if py_file.name.startswith("_"):
            continue

        try:
            module_classes = self._load_from_file(py_file)
            modules.update(module_classes)
        except Exception as e:
            logger.warning(f"无法从 {py_file} 加载模块: {e}")

    # 查找包
    for package_dir in module_dir.iterdir():
        if package_dir.is_dir() and (package_dir / "__init__.py").exists():
            try:
                module_classes = self._load_from_package(package_dir)
                modules.update(module_classes)
            except Exception as e:
                logger.error(f"无法从 {package_dir} 加载模块: {e}")

    return modules

ModuleLoader

Bases: ABC

模块加载器抽象基类.

Source code in src/symphra_modules/loader/base.py 
 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
 55
 56
 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
class ModuleLoader(ABC):
    """模块加载器抽象基类."""

    @abstractmethod
    def load(self, source: str) -> dict[str, type[ModuleInterface]]:
        """从源加载模块.

        Args:
            source: 模块源（目录路径、包名等）

        Returns:
            模块名到模块类的映射字典
        """

    @abstractmethod
    def discover(self, source: str) -> list[str]:
        """发现可用的模块.

        Args:
            source: 模块源

        Returns:
            模块名称列表
        """

    def _is_valid_module_class(self, obj: Any) -> bool:
        """检查对象是否是有效的模块类.

        Args:
            obj: 待检查的对象

        Returns:
            是否为有效的模块类
        """
        # 跳过抽象类和自身
        if obj is BaseModule or inspect.isabstract(obj):
            return False

        # 必须是类
        if not inspect.isclass(obj):
            return False

        # 检查必需的属性和方法
        required_attrs = ["metadata", "install", "start", "stop"]
        if not all(hasattr(obj, attr) for attr in required_attrs):
            return False

        # 检查 metadata 是否是 property
        return isinstance(getattr(type(obj), "metadata", None), property)

    def _validate_module_instance(self, obj: type[ModuleInterface]) -> bool:
        """验证模块实例是否有效.

        Args:
            obj: 模块类

        Returns:
            是否可以成功实例化并获取元数据
        """
        try:
            instance = obj()
            metadata = instance.metadata
            return isinstance(metadata, ModuleMetadata)
        except Exception as e:
            logger.warning(f"无法实例化模块类 {obj.__name__}: {e}")
            return False

    def _find_module_classes(self, module: Any) -> dict[str, type[ModuleInterface]]:
        """在模块中查找有效的模块类.

        Args:
            module: 要搜索的 Python 模块

        Returns:
            模块名到模块类的映射字典
        """
        module_classes: dict[str, type[ModuleInterface]] = {}

        # 获取模块中的所有类
        for name, obj in inspect.getmembers(module, inspect.isclass):
            # 基本有效性检查
            if not self._is_valid_module_class(obj):
                continue

            # 实例验证
            if self._validate_module_instance(obj):
                module_classes[name] = obj

        return module_classes
Functions
discover(source) abstractmethod

发现可用的模块.

参数：

名称 类型 描述 默认
source str

模块源

 必需

返回：

类型 描述
list[str]

模块名称列表
源代码位于： src/symphra_modules/loader/base.py 
27
28
29
30
31
32
33
34
35
36
@abstractmethod
def discover(self, source: str) -> list[str]:
    """发现可用的模块.

    Args:
        source: 模块源

    Returns:
        模块名称列表
    """
load(source) abstractmethod

从源加载模块.

参数：

名称 类型 描述 默认
source str

模块源（目录路径、包名等）

 必需

返回：

类型 描述
dict[str, type[ModuleInterface]]

模块名到模块类的映射字典
源代码位于： src/symphra_modules/loader/base.py 
16
17
18
19
20
21
22
23
24
25
@abstractmethod
def load(self, source: str) -> dict[str, type[ModuleInterface]]:
    """从源加载模块.

    Args:
        source: 模块源（目录路径、包名等）

    Returns:
        模块名到模块类的映射字典
    """

PackageLoader

Bases: ModuleLoader

从包名加载模块.

Source code in src/symphra_modules/loader/package.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
55
56
57
class PackageLoader(ModuleLoader):
    """从包名加载模块."""

    def load(self, source: str) -> dict[str, type[ModuleInterface]]:
        """从包名加载模块.

        Args:
            source: 包名（如 "my_package.modules"）

        Returns:
            模块类字典

        Raises:
            ModuleLoadError: 包未找到时抛出
        """
        try:
            module = importlib.import_module(source)
            return self._find_module_classes(module)
        except ImportError as e:
            raise ModuleLoadError(f"包未找到: {source}") from e

    def discover(self, source: str) -> list[str]:
        """发现包中的模块.

        Args:
            source: 包名

        Returns:
            模块名称列表
        """
        try:
            package = importlib.import_module(source)
            package_path = Path(package.__file__).parent if package.__file__ else Path.cwd() / source

            module_names: list[str] = []

            # 扫描包内的模块
            for item in package_path.iterdir():
                if item.is_file() and item.suffix == ".py" and not item.name.startswith("_"):
                    module_names.append(item.stem)
                elif item.is_dir() and (item / "__init__.py").exists():
                    module_names.append(item.name)

            return module_names

        except (ImportError, AttributeError):
            return []
Functions
discover(source)

发现包中的模块.

参数：

名称 类型 描述 默认
source str

包名

 必需

返回：

类型 描述
list[str]

模块名称列表
源代码位于： src/symphra_modules/loader/package.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def discover(self, source: str) -> list[str]:
    """发现包中的模块.

    Args:
        source: 包名

    Returns:
        模块名称列表
    """
    try:
        package = importlib.import_module(source)
        package_path = Path(package.__file__).parent if package.__file__ else Path.cwd() / source

        module_names: list[str] = []

        # 扫描包内的模块
        for item in package_path.iterdir():
            if item.is_file() and item.suffix == ".py" and not item.name.startswith("_"):
                module_names.append(item.stem)
            elif item.is_dir() and (item / "__init__.py").exists():
                module_names.append(item.name)

        return module_names

    except (ImportError, AttributeError):
        return []
load(source)

从包名加载模块.

参数：

名称 类型 描述 默认
source str

包名（如 "my_package.modules"）

 必需

返回：

类型 描述
dict[str, type[ModuleInterface]]

模块类字典

引发：

类型 描述
ModuleLoadError

包未找到时抛出
源代码位于： src/symphra_modules/loader/package.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def load(self, source: str) -> dict[str, type[ModuleInterface]]:
    """从包名加载模块.

    Args:
        source: 包名（如 "my_package.modules"）

    Returns:
        模块类字典

    Raises:
        ModuleLoadError: 包未找到时抛出
    """
    try:
        module = importlib.import_module(source)
        return self._find_module_classes(module)
    except ImportError as e:
        raise ModuleLoadError(f"包未找到: {source}") from e