graphlib.TopologicalSorter is a Python standard-library class for ordering dependencies in a directed acyclic graph. It solves a common production problem: a task may run only after all of its prerequisites have completed. Build systems, data pipelines, database migrations, report generation, deployment workflows, and file-processing applications all need this kind of ordering.
Instead of maintaining a fragile handwritten sequence, you describe which nodes depend on which predecessors. TopologicalSorter calculates a valid order and also provides an incremental API that can feed ready tasks to parallel workers without violating dependencies.
What topological ordering means
A topological order is a sequence in which every dependency appears before the item that depends on it. Imagine four steps: download data, validate it, transform it, and publish it. Validation depends on download, transformation depends on validation, and publication depends on transformation. The valid sequence is obvious in this small example, but larger graphs often contain several independent tasks that may be executed at the same time.
Topological sorting works only for graphs without cycles. If task A depends on B while B depends on A, no valid order exists. Python detects this condition and raises CycleError.
Your first static_order example
from graphlib import TopologicalSorter
graph = {
"publish": {"transform"},
"transform": {"validate"},
"validate": {"download"},
"download": set(),
}
order = list(TopologicalSorter(graph).static_order())
print(order)
The dictionary maps each task to its set of predecessors. static_order() prepares the graph, checks for cycles, and yields one valid sequence. For sequential scripts, this is the simplest interface.
Do not treat the relative position of independent nodes as a business guarantee. If two tasks do not depend on one another, either may appear first and the result can still be correct.
Building a graph with add
You can also construct the graph incrementally:
from graphlib import TopologicalSorter
ts = TopologicalSorter()
ts.add("download")
ts.add("validate", "download")
ts.add("transform", "validate")
ts.add("publish", "transform")
print(tuple(ts.static_order()))
add(node, *predecessors) accumulates dependencies. Calling add again for the same node adds more predecessors. This is useful when plugins, modules, or configuration files contribute separate pieces of one workflow.
Running tasks in parallel
The incremental API is designed for concurrent execution. First call prepare(). Then get_ready() returns every node whose dependencies are satisfied. When a worker finishes, call done() with that task. More nodes may then become ready.
from graphlib import TopologicalSorter
from concurrent.futures import ThreadPoolExecutor, wait, FIRST_COMPLETED
requirements = {
"extract_customers": set(),
"extract_orders": set(),
"join": {"extract_customers", "extract_orders"},
"report": {"join"},
}
def run(name):
print("running", name)
return name
ts = TopologicalSorter(requirements)
ts.prepare()
with ThreadPoolExecutor(max_workers=4) as pool:
futures = {}
while ts.is_active() or futures:
for task in ts.get_ready():
futures[pool.submit(run, task)] = task
if not futures:
break
completed, _ = wait(futures, return_when=FIRST_COMPLETED)
for future in completed:
task = futures.pop(future)
future.result()
ts.done(task)
This pattern allows parallelism only among ready tasks. Calling future.result() also propagates worker exceptions. In a production system, define the failure policy explicitly: stop the entire graph, retry the failed task, block descendants, or perform compensation.
prepare, get_ready, done, and is_active
prepare() finalizes graph preparation and detects cycles. Once execution begins, the graph should be treated as fixed. get_ready() may return several nodes, and every returned node should later be reported through done() after successful completion.
Never mark a task done before its real work finishes. That releases dependent nodes too early and turns a correct dependency graph into an incorrect execution. Do not call done() twice for the same node either.
is_active() tells you whether useful work remains or tasks are still outstanding. It is convenient for controlling scheduler loops, especially when tasks are sent to an external queue.
Detecting cycles
from graphlib import TopologicalSorter, CycleError
graph = {
"a": {"b"},
"b": {"c"},
"c": {"a"},
}
try:
list(TopologicalSorter(graph).static_order())
except CycleError as error:
print("cycle detected", error.args)
Cycles usually indicate a modeling or configuration mistake. The exception may expose nodes involved in the cycle, but critical logic should not depend on one exact error representation. A better diagnostic records the loaded dependency edges and shows users a readable chain.
Normalizing configuration
Dependencies loaded from YAML, JSON, or a database may contain duplicate names, missing nodes, and empty strings. Normalize input before creating the sorter.
def normalize(config):
graph = {}
for name, dependencies in config.items():
name = name.strip()
if not name:
raise ValueError("task without a name")
graph[name] = {d.strip() for d in dependencies if d.strip()}
return graph
TopologicalSorter accepts predecessors that never appear as dictionary keys and automatically includes them as nodes. This can be convenient, but it may hide a typo. Strict systems should verify every dependency against a known task catalog.
A file pipeline example
from graphlib import TopologicalSorter
pipeline = {
"archive": {"make_csv", "make_pdf"},
"send": {"archive"},
"make_csv": {"query"},
"make_pdf": {"query", "load_template"},
"query": set(),
"load_template": set(),
}
for step in TopologicalSorter(pipeline).static_order():
print(step)
query and load_template can run first, possibly in parallel. make_pdf waits for both, while make_csv depends only on the query. archive waits for both generated artifacts, and send finishes the flow.
Managing outputs
The sorter organizes nodes but does not store task results. Keep a result dictionary or an external store:
results = {}
def execute(task):
if task == "query":
results[task] = [1, 2, 3]
elif task == "make_csv":
results[task] = create_csv(results["query"])
With threads, protect shared mutable structures when writes may overlap. With processes or distributed workers, use a database, cache, message broker, or object store. Associate outputs with a run identifier so concurrent executions cannot mix data.
Idempotency and recovery
A reliable orchestrator must survive restarts. Make tasks idempotent whenever possible: running a task again should produce the same final state. Record states such as pending, running, completed, and failed. On recovery, rebuild the graph and consider a node complete only after its output has been verified.
TopologicalSorter does not persist state and is not a replacement for Airflow, Prefect, Celery, or a dedicated workflow engine. It is a lightweight building block for applications that need local dependency control or want to implement a custom scheduler.
Retries and failure propagation
Retries should belong to the task-execution layer, not to the dependency graph itself. Limit retry counts, use exponential backoff for transient failures, and avoid retrying permanent validation errors. Descendants of a failed node should remain blocked unless the failure is explicitly ignored or an alternative dependency path exists.
Store the original exception, attempt count, start time, and finish time. These records make incidents understandable and help distinguish slow dependencies from actual scheduler problems.
Memory and scale
The graph is kept in memory. Thousands or tens of thousands of nodes may be perfectly reasonable, but measure your workload. Huge dynamic graphs may require partitioning or a specialized system. Topological sorting also does not reduce the cost of the tasks; it only determines when each one is allowed to start.
Prefer immutable hashable node identifiers such as strings, integers, enums, or tuples. Nodes are used as dictionary and set members, so mutable values are unsuitable.
Testing dependency graphs
Test more than the happy path. Include independent branches, a shared prerequisite, a node with several predecessors, an unknown dependency, and at least one deliberate cycle. For concurrent execution, verify that descendants never start before all predecessors have completed.
A useful unit test checks relationships instead of one exact total order. For every edge dependency -> task, assert that the dependency index is lower than the task index. This keeps tests correct even when independent nodes change positions.
Best practices
Use stable names, validate unknown dependencies, detect cycles before expensive work starts, limit worker concurrency, propagate failures, record durations and outputs, and never confuse “ready” with “completed.” Important workflows should persist execution events and include correlation identifiers in logs.
Related Academify guides include Python dictionaries, Python sets, Python threading, and Python concurrent.futures. External references include the official graphlib documentation and the official concurrent.futures documentation.
Conclusion
graphlib.TopologicalSorter turns a network of dependencies into an executable order and provides a safe foundation for controlled parallelism. For small and medium workflows, it replaces manual sequencing, detects cycles, and makes task contracts explicit. Combined with validation, persistence, idempotency, and careful failure handling, it becomes a practical component for building reliable pipelines with Python’s standard library.
