功能说明
这段代码把“任务到前置依赖”的映射转换成可执行的先后顺序。它适合构建系统、迁移脚本或初始化流程:只有当一个节点的所有前置依赖都完成后,该节点才会进入就绪队列。如果图中存在循环,函数不会静默返回残缺顺序,而是抛出包含阻塞节点的错误。
实现使用 Kahn 算法,并用 dependents 反向索引记录“某个依赖被哪些节点需要”,从而避免反复扫描整张图。
关键代码
build_order 先为每个节点统计入度:一条 node -> dep 的记录会让 dep 的入度保持零,同时让 node 的入度加一。所有入度为零的节点按名称排序后放入 deque。
取出节点后,把它加入结果,并遍历其 dependents,将子节点入度减一;减到零时追加到队列。循环结束后,如果结果长度小于节点总数,说明仍有节点入度大于零,代码会收集这些节点并抛出 ValueError。
使用注意
依赖列表中的重复项会让入度重复累加,但 dependents 也会重复记录,因此最终仍能抵消;不过最好在输入阶段去重。检测到的“阻塞节点”可能包含环下游的节点,不一定只是环本身。函数不校验节点名是否存在,未知依赖会被当作独立节点加入。
排序结果只保证拓扑正确,不保证全局字典序最小;如需稳定输出,可改用优先队列。图很大时应关注内存占用,因为需要保存完整邻接表。
验证方式
示例图会输出 db -> docs -> auth -> api -> app,代码断言 db 在 api 之前、api 在 app 之前。随后构造 a 依赖 b、b 又依赖 a 的环,捕获异常并打印 dependency cycle among: a, b。可以再加入自依赖 x: ["x"] 验证同样会被拒绝。
代码示例
from __future__ import annotations
from collections import defaultdict, deque
def build_order(prerequisites: dict[str, list[str]]) -> list[str]:
indegree = {node: 0 for node in prerequisites}
dependents: dict[str, list[str]] = defaultdict(list)
for node, deps in prerequisites.items():
for dep in deps:
indegree.setdefault(dep, 0)
dependents[dep].append(node)
indegree[node] += 1
ready = deque(sorted(node for node, degree in indegree.items() if degree == 0))
order: list[str] = []
while ready:
node = ready.popleft()
order.append(node)
for child in sorted(dependents[node]):
indegree[child] -= 1
if indegree[child] == 0:
ready.append(child)
if len(order) != len(indegree):
blocked = sorted(node for node, degree in indegree.items() if degree > 0)
raise ValueError(f"dependency cycle among: {', '.join(blocked)}")
return order
def main() -> None:
graph = {
"app": ["api", "db"],
"api": ["db", "auth"],
"auth": ["db"],
"db": [],
"docs": [],
}
order = build_order(graph)
print(" -> ".join(order))
assert order.index("db") < order.index("api") < order.index("app")
try:
build_order({"a": ["b"], "b": ["a"]})
except ValueError as exc:
print(f"cycle: {exc}")
if __name__ == "__main__":
main()