Many software workflows contain tasks that cannot start until other tasks have finished. A package depends on libraries, a deployment depends on tests, a database migration depends on the previous migration, and a data pipeline must extract information before transforming it. These relationships form a directed dependency graph. The standard-library Python graphlib module provides TopologicalSorter, a focused tool for producing a valid execution order, finding work that can run concurrently, and detecting circular dependencies.
This guide explains how to model predecessors, generate a static order, coordinate parallel workers, diagnose cycles, validate task names, and test a dependency graph. The topic complements practical guides about Python contextvars, priority queues with heapq, cached_property, and singledispatch.
What is topological sorting?
A topological order is a linear sequence of vertices in a directed graph where every prerequisite appears before the item that depends on it. If deploy depends on test, the test step must occur earlier in the sequence. A graph can have several valid topological orders when independent tasks are available at the same time.
A valid order exists only for a directed acyclic graph. A cycle appears when task A depends on B, B depends on C, and C eventually depends on A. No task in that loop can start, because each one waits for another member of the same cycle. TopologicalSorter reports this problem with CycleError.
Representing dependencies
The constructor accepts a dictionary that maps each node to an iterable of its predecessors. This direction is important: the values are requirements, not successors.
from graphlib import TopologicalSorter
dependencies = {
"deploy": {"test"},
"test": {"build"},
"build": {"install"},
"install": set(),
}
order = list(TopologicalSorter(dependencies).static_order())
print(order)The resulting sequence respects every relationship. The exact output may differ when two nodes do not depend on each other, and applications should not rely on an arbitrary tie order.
Using static_order
For a sequential workflow, static_order() is the simplest interface. It prepares the graph, checks for cycles, and returns an iterator over the nodes:
from graphlib import TopologicalSorter
steps = {
"send_email": {"generate_report"},
"generate_report": {"query_database", "load_config"},
"query_database": {"load_config"},
"load_config": set(),
}
for step in TopologicalSorter(steps).static_order():
print(f"Running: {step}")This approach is ideal for validation tools, small scripts, migration plans, and workflows that intentionally execute one task at a time.
Adding nodes incrementally
You can also create an empty sorter and register dependencies with add(). The first argument is the node and the remaining arguments are its predecessors:
from graphlib import TopologicalSorter
sorter = TopologicalSorter()
sorter.add("build", "lint", "test")
sorter.add("deploy", "build")
sorter.add("lint")
sorter.add("test", "install")
sorter.add("install")
print(list(sorter.static_order()))Repeated calls for the same node accumulate prerequisites. This is useful when plugins or modules contribute parts of a larger pipeline. Once preparation begins, however, the graph should no longer be modified.
Incremental and parallel processing
The most powerful use of TopologicalSorter is coordinating tasks that can run concurrently. The relevant methods are prepare(), get_ready(), done(), and is_active().
from graphlib import TopologicalSorter
steps = {
"deploy": {"tests", "frontend_build"},
"tests": {"install"},
"frontend_build": {"install"},
"install": set(),
}
sorter = TopologicalSorter(steps)
sorter.prepare()
while sorter.is_active():
ready = sorter.get_ready()
for task in ready:
print("Running", task)
# run the task and wait for success
sorter.done(task)After install is marked complete, both tests and frontend_build become ready. They may be submitted to threads, processes, containers, or remote workers. Calling done() releases downstream nodes only after the application confirms completion.
Using ThreadPoolExecutor
For input/output-heavy steps, a thread pool can process independent nodes concurrently:
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
from graphlib import TopologicalSorter
import time
steps = {
"publish": {"test", "document"},
"test": {"compile"},
"document": {"compile"},
"compile": {"download"},
"download": set(),
}
def run(name):
print("start", name)
time.sleep(0.2)
print("finish", name)
return name
sorter = TopologicalSorter(steps)
sorter.prepare()
futures = {}
with ThreadPoolExecutor(max_workers=3) as executor:
while sorter.is_active():
for task in sorter.get_ready():
future = executor.submit(run, task)
futures[future] = task
completed, _ = wait(futures, return_when=FIRST_COMPLETED)
for future in completed:
task = futures.pop(future)
future.result()
sorter.done(task)A production implementation must decide what happens after a failure. Do not call done() when a prerequisite failed unless the workflow explicitly allows downstream execution. Common policies include stopping the graph, retrying the task, marking dependents as skipped, or recording a partial result.
Detecting circular dependencies
Both prepare() and iteration through static_order() can raise CycleError. The exception arguments include information about one detected cycle:
from graphlib import CycleError, TopologicalSorter
graph = {
"a": {"c"},
"b": {"a"},
"c": {"b"},
}
try:
print(list(TopologicalSorter(graph).static_order()))
except CycleError as error:
print("Circular dependency:", error.args)A sorter may still expose nodes outside the cycle before progress becomes impossible. Diagnostic tools can use that behavior to report both the valid portion and the blocked component.
Implicit predecessor nodes
If a predecessor is referenced but does not appear as a dictionary key, graphlib automatically adds it as a node with no prerequisites. This is convenient, but it can hide spelling mistakes. For example, configuration and configuraton become separate nodes.
Validate all referenced identifiers against an approved catalog:
catalog = {"config", "extract", "transform", "load"}
dependencies = {
"transform": {"extract"},
"load": {"transform", "config"},
}
referenced = set(dependencies)
for predecessors in dependencies.values():
referenced.update(predecessors)
unknown = referenced - catalog
if unknown:
raise ValueError(f"Unknown tasks: {unknown}")Using objects as nodes
Nodes must be hashable, but they do not have to be strings. Enums, tuples, integers, and immutable data classes can carry richer metadata:
from dataclasses import dataclass
from graphlib import TopologicalSorter
@dataclass(frozen=True)
class Task:
name: str
team: str
extract = Task("extract", "data")
validate = Task("validate", "quality")
publish = Task("publish", "platform")
graph = {
validate: {extract},
publish: {validate},
}
for task in TopologicalSorter(graph).static_order():
print(task.name, task.team)Avoid mutable objects whose equality or hash may change after insertion. Changing hashed state can make dictionary and set behavior unpredictable.
Practical applications
Topological sorting is useful for:
- package and plugin dependency resolution;
- data engineering and machine-learning pipelines;
- database migration ordering;
- module compilation;
- course prerequisite planning;
- spreadsheet and report generation;
- service startup sequencing;
- build systems and deployment workflows.
The module determines readiness and order. It does not provide persistence, distributed queues, resource quotas, automatic retries, schedules, or monitoring. A larger system can use it as the planning core while another component executes and records tasks.
Testing a dependency graph
When multiple valid orders exist, testing one exact list makes the test unnecessarily fragile. Instead, verify the defining property: every predecessor must appear before the dependent node.
from graphlib import TopologicalSorter
def assert_valid_order(graph):
order = list(TopologicalSorter(graph).static_order())
position = {node: index for index, node in enumerate(order)}
for node, predecessors in graph.items():
for predecessor in predecessors:
assert position[predecessor] < position[node]
assert_valid_order({
"deploy": {"test", "build"},
"test": {"install"},
"build": {"install"},
"install": set(),
})Add tests for an empty graph, an isolated node, several independent roots, direct and indirect cycles, duplicate registrations, unknown identifiers, and worker failures.
Common mistakes
- Providing successors instead of predecessors.
- Assuming independent nodes always have the same relative order.
- Forgetting to call
done()during incremental processing. - Calling
done()before work actually succeeds. - Trying to modify the graph after preparation.
- Ignoring worker exceptions and releasing dependents.
- Using inconsistent names that create unintended nodes.
- Expecting graphlib to execute tasks automatically.
Recommended practices
- Use stable, immutable node identifiers.
- Validate unknown nodes before preparation.
- Document that the mapping points from nodes to predecessors.
- Separate planning, execution, and persistence.
- Define failure and retry policies before adding concurrency.
- Limit parallelism according to CPU, memory, and external services.
- Record task start, finish, duration, and outcome.
- Test graph properties rather than arbitrary tie ordering.
Conclusion
Python graphlib turns dependency relationships into a valid execution plan without third-party packages. static_order() handles straightforward sequential cases, while prepare(), get_ready(), done(), and is_active() support concurrent scheduling.
The essential design work is modeling predecessors accurately and treating cycles as configuration errors. TopologicalSorter is not a complete workflow orchestrator, but it is a reliable foundation for installers, build systems, migrations, data pipelines, and any process governed by prerequisites.
For version-specific behavior, consult the official graphlib documentation and the concurrent.futures documentation.
