Python graphlib TopologicalSorter Dependency Resolution

This article explains how Python's graphlib.TopologicalSorter class resolves dependency graphs, offering an overview of its core functionality, cycle detection capabilities, and support for both static and dynamic task scheduling. By leveraging directed acyclic graphs (DAGs), developers can systematically determine the exact order required to execute tasks, build packages, or evaluate prerequisites without manually implementing complex sorting algorithms.

Understanding Topological Sorting and graphlib

Introduced in Python 3.9, the graphlib module provides tools to implement topological sorts of directed acyclic graphs (DAGs). In a dependency graph, nodes represent tasks or items, and directed edges represent prerequisites. A topological sort arranges these nodes linearly such that every prerequisite appears before the node that depends on it.

The graphlib.TopologicalSorter class automates this ordering, eliminating the need to write custom graph traversal algorithms like Kahn's algorithm or depth-first search.

Static Dependency Resolution

When all dependencies are known upfront and tasks are executed sequentially, TopologicalSorter can generate a complete execution order using the static_order() method.

from graphlib import TopologicalSorter

# Define dependencies: Node -> set of prerequisites
dependencies = {
    "deploy": {"build", "test"},
    "test": {"compile"},
    "build": {"compile"},
    "compile": set(),
}

sorter = TopologicalSorter(dependencies)
execution_order = tuple(sorter.static_order())
print(execution_order)
# Output: ('compile', 'build', 'test', 'deploy') (or 'test' before 'build')

In this mode, the sorter reads the mapping where keys depend on values, then outputs a flattened tuple containing a valid sequence of operations.

Dynamic and Parallel Task Execution

TopologicalSorter is designed to handle dynamic workflows where tasks finish asynchronously or run in parallel. Instead of pre-calculating the entire sequence, you can control the sorting lifecycle using three methods:

  1. prepare(): Finalizes the graph setup and checks for cycles. Must be called before querying tasks.
  2. get_ready(): Returns a tuple of all nodes whose prerequisites have finished and are ready to be processed.
  3. done(*nodes): Informs the sorter that the specified nodes have completed, unlocking downstream nodes that were waiting on them.
from graphlib import TopologicalSorter

graph = {
    "database": set(),
    "backend": {"database"},
    "frontend": set(),
    "app": {"backend", "frontend"},
}

ts = TopologicalSorter(graph)
ts.prepare()

while ts.is_active():
    ready_nodes = ts.get_ready()
    for node in ready_nodes:
        # Simulate processing the node
        print(f"Executing: {node}")
        ts.done(node)

This pattern makes the class suitable for orchestrating task runners, workflow pipelines, and concurrent job queues using asyncio or concurrent.futures.

Automatic Cycle Detection

A dependency graph cannot be resolved if circular dependencies exist (for example, A depends on B, and B depends on A). The TopologicalSorter automatically detects cycles and raises a graphlib.CycleError.

from graphlib import TopologicalSorter, CycleError

cyclic_graph = {
    "A": {"B"},
    "B": {"C"},
    "C": {"A"}
}

ts = TopologicalSorter(cyclic_graph)
try:
    ts.prepare()
except CycleError as exc:
    print(f"Cycle detected: {exc}")

The exception details the cycle path, allowing applications to pinpoint and report configuration errors or deadlocks before runtime execution begins.