跳转至

依赖解析

resolver

依赖解析器.

Classes

DependencyGraph

依赖关系图.

Source code in src/symphra_modules/resolver/dependency.py 
 9
10
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
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class DependencyGraph:
    """依赖关系图."""

    def __init__(self) -> None:
        """初始化依赖图."""
        self._graph: dict[str, set[str]] = defaultdict(set)  # 模块 -> 依赖的模块集合
        self._reverse_graph: dict[str, set[str]] = defaultdict(set)  # 模块 -> 依赖它的模块集合
        self._modules: dict[str, ModuleMetadata] = {}

    def add_module(self, metadata: ModuleMetadata) -> None:
        """添加模块到依赖图.

        Args:
            metadata: 模块元数据
        """
        name = metadata.name
        self._modules[name] = metadata

        # 初始化空依赖集合（即使没有依赖）
        if name not in self._graph:
            self._graph[name] = set()

        # 添加依赖关系
        for dep in metadata.dependencies:
            self._graph[name].add(dep)
            self._reverse_graph[dep].add(name)

    def get_dependencies(self, name: str) -> set[str]:
        """获取模块的直接依赖.

        Args:
            name: 模块名称

        Returns:
            依赖的模块名称集合
        """
        return self._graph.get(name, set())

    def get_dependents(self, name: str) -> set[str]:
        """获取依赖该模块的模块.

        Args:
            name: 模块名称

        Returns:
            依赖该模块的模块名称集合
        """
        return self._reverse_graph.get(name, set())

    def has_module(self, name: str) -> bool:
        """检查模块是否存在.

        Args:
            name: 模块名称

        Returns:
            是否存在
        """
        return name in self._modules

    def get_all_modules(self) -> list[str]:
        """获取所有模块名称.

        Returns:
            模块名称列表
        """
        return list(self._modules.keys())
Functions
__init__()

初始化依赖图.

源代码位于： src/symphra_modules/resolver/dependency.py 
12
13
14
15
16
def __init__(self) -> None:
    """初始化依赖图."""
    self._graph: dict[str, set[str]] = defaultdict(set)  # 模块 -> 依赖的模块集合
    self._reverse_graph: dict[str, set[str]] = defaultdict(set)  # 模块 -> 依赖它的模块集合
    self._modules: dict[str, ModuleMetadata] = {}
add_module(metadata)

添加模块到依赖图.

参数：

名称 类型 描述 默认
metadata ModuleMetadata

模块元数据

 必需
源代码位于： src/symphra_modules/resolver/dependency.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
def add_module(self, metadata: ModuleMetadata) -> None:
    """添加模块到依赖图.

    Args:
        metadata: 模块元数据
    """
    name = metadata.name
    self._modules[name] = metadata

    # 初始化空依赖集合（即使没有依赖）
    if name not in self._graph:
        self._graph[name] = set()

    # 添加依赖关系
    for dep in metadata.dependencies:
        self._graph[name].add(dep)
        self._reverse_graph[dep].add(name)
get_all_modules()

获取所有模块名称.

返回：

类型 描述
list[str]

模块名称列表
源代码位于： src/symphra_modules/resolver/dependency.py 
69
70
71
72
73
74
75
def get_all_modules(self) -> list[str]:
    """获取所有模块名称.

    Returns:
        模块名称列表
    """
    return list(self._modules.keys())
get_dependencies(name)

获取模块的直接依赖.

参数：

名称 类型 描述 默认
name str

模块名称

 必需

返回：

类型 描述
set[str]

依赖的模块名称集合
源代码位于： src/symphra_modules/resolver/dependency.py 
36
37
38
39
40
41
42
43
44
45
def get_dependencies(self, name: str) -> set[str]:
    """获取模块的直接依赖.

    Args:
        name: 模块名称

    Returns:
        依赖的模块名称集合
    """
    return self._graph.get(name, set())
get_dependents(name)

获取依赖该模块的模块.

参数：

名称 类型 描述 默认
name str

模块名称

 必需

返回：

类型 描述
set[str]

依赖该模块的模块名称集合
源代码位于： src/symphra_modules/resolver/dependency.py 
47
48
49
50
51
52
53
54
55
56
def get_dependents(self, name: str) -> set[str]:
    """获取依赖该模块的模块.

    Args:
        name: 模块名称

    Returns:
        依赖该模块的模块名称集合
    """
    return self._reverse_graph.get(name, set())
has_module(name)

检查模块是否存在.

参数：

名称 类型 描述 默认
name str

模块名称

 必需

返回：

类型 描述
bool

是否存在
源代码位于： src/symphra_modules/resolver/dependency.py 
58
59
60
61
62
63
64
65
66
67
def has_module(self, name: str) -> bool:
    """检查模块是否存在.

    Args:
        name: 模块名称

    Returns:
        是否存在
    """
    return name in self._modules

DependencyResolver

依赖解析器 - 使用拓扑排序解析模块加载顺序.

Source code in src/symphra_modules/resolver/dependency.py 
 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
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
class DependencyResolver:
    """依赖解析器 - 使用拓扑排序解析模块加载顺序."""

    def __init__(self) -> None:
        """初始化解析器."""
        self.graph = DependencyGraph()

    def add_module(self, metadata: ModuleMetadata) -> None:
        """添加模块.

        Args:
            metadata: 模块元数据
        """
        self.graph.add_module(metadata)

    def resolve(self) -> list[str]:
        """解析模块加载顺序（拓扑排序）.

        使用 Kahn 算法进行拓扑排序。

        Returns:
            按依赖顺序排列的模块名称列表

        Raises:
            ModuleDependencyError: 存在循环依赖或缺失依赖时抛出
        """
        # 检查缺失的依赖
        self._check_missing_dependencies()

        # 计算每个模块的入度（被依赖的次数）
        in_degree: dict[str, int] = defaultdict(int)
        for module in self.graph.get_all_modules():
            in_degree[module] = 0

        for module in self.graph.get_all_modules():
            for _dep in self.graph.get_dependencies(module):
                in_degree[module] += 1

        # 找到所有入度为 0 的模块（没有依赖的模块）
        queue: deque[str] = deque()
        for module in self.graph.get_all_modules():
            if in_degree[module] == 0:
                queue.append(module)

        result: list[str] = []

        while queue:
            # 取出一个入度为 0 的模块
            current = queue.popleft()
            result.append(current)

            # 对于所有依赖当前模块的模块，减少其入度
            for dependent in self.graph.get_dependents(current):
                in_degree[dependent] -= 1
                if in_degree[dependent] == 0:
                    queue.append(dependent)

        # 如果结果数量不等于模块总数，说明存在循环依赖
        if len(result) != len(self.graph.get_all_modules()):
            circular = self._detect_circular_dependencies(in_degree)
            raise ModuleDependencyError(
                "检测到循环依赖",
                circular_dependencies=circular,
            )

        return result

    def _check_missing_dependencies(self) -> None:
        """检查缺失的依赖.

        Raises:
            ModuleDependencyError: 存在缺失依赖时抛出
        """
        missing_deps: dict[str, list[str]] = {}

        for module in self.graph.get_all_modules():
            missing: list[str] = []
            for dep in self.graph.get_dependencies(module):
                if not self.graph.has_module(dep):
                    missing.append(dep)

            if missing:
                missing_deps[module] = missing

        if missing_deps:
            # 格式化错误消息
            error_details = []
            for module, deps in missing_deps.items():
                error_details.append(f"{module}: {', '.join(deps)}")

            raise ModuleDependencyError(
                f"缺失依赖: {'; '.join(error_details)}",
                missing_dependencies=list(missing_deps.keys()),
            )

    def _detect_circular_dependencies(self, in_degree: dict[str, int]) -> list[str]:
        """检测循环依赖.

        Args:
            in_degree: 模块入度映射

        Returns:
            参与循环依赖的模块列表
        """
        circular: list[str] = []
        for module, degree in in_degree.items():
            if degree > 0:
                circular.append(module)
        return circular

    def get_load_order_for_module(self, name: str) -> list[str]:
        """获取加载特定模块所需的顺序（包括其所有依赖）.

        Args:
            name: 模块名称

        Returns:
            按依赖顺序排列的模块名称列表（包含该模块自身）

        Raises:
            ModuleDependencyError: 模块不存在或依赖有问题时抛出
        """
        if not self.graph.has_module(name):
            raise ModuleDependencyError(f"模块不存在: {name}", module_name=name)

        # 使用 DFS 收集所有依赖
        visited: set[str] = set()
        stack: list[str] = []
        visiting: set[str] = set()  # 用于检测循环依赖

        def dfs(module: str) -> None:
            if module in visited:
                return
            if module in visiting:
                # 检测到循环依赖
                raise ModuleDependencyError(
                    f"检测到循环依赖，涉及模块: {module}",
                    module_name=module,
                    circular_dependencies=[module],
                )

            visiting.add(module)

            # 先处理依赖
            for dep in self.graph.get_dependencies(module):
                if not self.graph.has_module(dep):
                    raise ModuleDependencyError(
                        f"缺失依赖: {module} 依赖 {dep}",
                        module_name=module,
                        missing_dependencies=[dep],
                    )
                dfs(dep)

            visiting.remove(module)
            visited.add(module)
            stack.append(module)

        dfs(name)
        return stack
Functions
__init__()

初始化解析器.

源代码位于： src/symphra_modules/resolver/dependency.py 
81
82
83
def __init__(self) -> None:
    """初始化解析器."""
    self.graph = DependencyGraph()
add_module(metadata)

添加模块.

参数：

名称 类型 描述 默认
metadata ModuleMetadata

模块元数据

 必需
源代码位于： src/symphra_modules/resolver/dependency.py 
85
86
87
88
89
90
91
def add_module(self, metadata: ModuleMetadata) -> None:
    """添加模块.

    Args:
        metadata: 模块元数据
    """
    self.graph.add_module(metadata)
get_load_order_for_module(name)

获取加载特定模块所需的顺序（包括其所有依赖）.

参数：

名称 类型 描述 默认
name str

模块名称

 必需

返回：

类型 描述
list[str]

按依赖顺序排列的模块名称列表（包含该模块自身）

引发：

类型 描述
ModuleDependencyError

模块不存在或依赖有问题时抛出
源代码位于： src/symphra_modules/resolver/dependency.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
def get_load_order_for_module(self, name: str) -> list[str]:
    """获取加载特定模块所需的顺序（包括其所有依赖）.

    Args:
        name: 模块名称

    Returns:
        按依赖顺序排列的模块名称列表（包含该模块自身）

    Raises:
        ModuleDependencyError: 模块不存在或依赖有问题时抛出
    """
    if not self.graph.has_module(name):
        raise ModuleDependencyError(f"模块不存在: {name}", module_name=name)

    # 使用 DFS 收集所有依赖
    visited: set[str] = set()
    stack: list[str] = []
    visiting: set[str] = set()  # 用于检测循环依赖

    def dfs(module: str) -> None:
        if module in visited:
            return
        if module in visiting:
            # 检测到循环依赖
            raise ModuleDependencyError(
                f"检测到循环依赖，涉及模块: {module}",
                module_name=module,
                circular_dependencies=[module],
            )

        visiting.add(module)

        # 先处理依赖
        for dep in self.graph.get_dependencies(module):
            if not self.graph.has_module(dep):
                raise ModuleDependencyError(
                    f"缺失依赖: {module} 依赖 {dep}",
                    module_name=module,
                    missing_dependencies=[dep],
                )
            dfs(dep)

        visiting.remove(module)
        visited.add(module)
        stack.append(module)

    dfs(name)
    return stack
resolve()

解析模块加载顺序（拓扑排序）.

使用 Kahn 算法进行拓扑排序。

返回：

类型 描述
list[str]

按依赖顺序排列的模块名称列表

引发：

类型 描述
ModuleDependencyError

存在循环依赖或缺失依赖时抛出
源代码位于： src/symphra_modules/resolver/dependency.py 
 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
def resolve(self) -> list[str]:
    """解析模块加载顺序（拓扑排序）.

    使用 Kahn 算法进行拓扑排序。

    Returns:
        按依赖顺序排列的模块名称列表

    Raises:
        ModuleDependencyError: 存在循环依赖或缺失依赖时抛出
    """
    # 检查缺失的依赖
    self._check_missing_dependencies()

    # 计算每个模块的入度（被依赖的次数）
    in_degree: dict[str, int] = defaultdict(int)
    for module in self.graph.get_all_modules():
        in_degree[module] = 0

    for module in self.graph.get_all_modules():
        for _dep in self.graph.get_dependencies(module):
            in_degree[module] += 1

    # 找到所有入度为 0 的模块（没有依赖的模块）
    queue: deque[str] = deque()
    for module in self.graph.get_all_modules():
        if in_degree[module] == 0:
            queue.append(module)

    result: list[str] = []

    while queue:
        # 取出一个入度为 0 的模块
        current = queue.popleft()
        result.append(current)

        # 对于所有依赖当前模块的模块，减少其入度
        for dependent in self.graph.get_dependents(current):
            in_degree[dependent] -= 1
            if in_degree[dependent] == 0:
                queue.append(dependent)

    # 如果结果数量不等于模块总数，说明存在循环依赖
    if len(result) != len(self.graph.get_all_modules()):
        circular = self._detect_circular_dependencies(in_degree)
        raise ModuleDependencyError(
            "检测到循环依赖",
            circular_dependencies=circular,
        )

    return result