How to Topologically Sort a DAG in Python
Compute a valid execution order for tasks with dependencies using Kahn's algorithm in Python.
Python code
40 linesfrom collections import defaultdict, deque
def topological_order(dependencies):
graph = defaultdict(list)
in_degree = defaultdict(int)
tasks = set(dependencies.keys())
for task, depends_on in dependencies.items():
for d in depends_on:
graph[d].append(task)
in_degree[task] += 1
tasks.add(d)
queue = deque([t for t in tasks if in_degree[t] == 0])
order = []
while queue:
current = queue.popleft()
order.append(current)
for neighbor in graph[current]:
in_degree[neighbor] -= 1
if in_degree[neighbor] == 0:
queue.append(neighbor)
if len(order) != len(tasks):
raise ValueError("Cycle detected in dependencies")
return order
if __name__ == "__main__":
deps = {
"build": ["compile", "test"],
"compile": ["parse"],
"test": ["compile"],
"deploy": ["build"],
"parse": [],
}
print(topological_order(deps))
Output
['parse', 'compile', 'test', 'build', 'deploy']
How it works
The graph is built from a dependencies dictionary where each key points to a list of tasks it depends on. We track in-degree (number of unresolved dependencies) per task. Tasks with in-degree zero are queued and processed, decrementing neighbors' in-degrees as they become ready. If all tasks are visited, the order is valid; otherwise a cycle exists and an exception is raised.
Common mistakes
- Forgetting to include all dependent tasks in the `tasks` set, leading to missing nodes.
- Not checking for cycles — a partial order may silently omit some tasks.
- Assuming the order is unique; multiple valid orders exist.
Variations
- Use a recursive DFS with visited marks for topological sorting.
- Use the `toposort` third-party library for a simpler API.
Real-world use cases
- Determining build order for a CI pipeline where steps depend on prior artifacts.
- Scheduling data processing stages where one ETL job consumes output of another.
- Resolving package installation order based on dependencies between software packages.
Sponsored
More from Data pipelines & processing
- Add a UUID Surrogate Key to Each Row in a CSV with Python easy
- Attach Source File Metadata to Records in Python easy
- Build a Python Utility That Detects Duplicate Records Across Multiple Excel Sheets medium
- Check Null Rate Threshold in PySpark DataFrame medium
- Count Records Processed per Category in Python easy
- Create Data Helper Functions in Python for Beginners easy
Keep learning
Related tutorials and quizzes for this topic.