Algo Topological Sort

Topological Sort

Linear ordering of vertices in a DAG such that for every edge u->v, u appears before v. Both algorithms below return None if the graph has a cycle.

When to Use

• Determining valid execution order for a pipeline with step dependencies (e.g. QC -> trim -> align -> count -> DE analysis).
• Ordering a gene regulatory network or signal transduction cascade (upstream regulator before downstream target).
• Scheduling metabolic pathway reactions that must occur in enzymatic sequence.
• Detecting whether a dependency graph (build system, task DAG, workflow spec) contains a cycle before executing it.
• Computing the critical path / minimum makespan of a weighted task DAG under unlimited parallelism.

Version Compatibility

Pure Python standard library only (collections.defaultdict, collections.deque). Works on Python >= 3.7 (relies on dict insertion order and f-strings); no third-party dependencies.

Prerequisites

• Comfortable with graph representations (adjacency lists) — see algo-graph-representations.
• Basic DFS/BFS traversal — see algo-bfs-dfs.
• No packages to install beyond the standard library.

Building the Graph

Goal: represent a directed graph with an adjacency list. Approach: defaultdict(list) for edges plus a set of all vertices seen so far.

from collections import defaultdict, deque

class DirectedGraph:
    """Adjacency-list directed graph used by all topo-sort routines below."""
    def __init__(self):
        self.adj = defaultdict(list)
        self.vertices = set()

    def add_edge(self, u, v):
        self.adj[u].append(v)
        self.vertices.add(u)
        self.vertices.add(v)

DFS-Based Sort — O(V + E)

Goal: produce a topological ordering, or detect that none exists. Approach: DFS with a 3-color scheme (WHITE=unvisited, GRAY=on current recursion stack, BLACK=finished). A GRAY-to-GRAY edge is a back edge, i.e. a cycle. Push each vertex onto result on finish, then reverse.

def topological_sort_dfs(graph):
    """Topological sort via DFS. Returns ordering, or None if graph has a cycle."""
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {v: WHITE for v in graph.vertices}
    result = []

    def dfs(v):
        color[v] = GRAY
        for neighbor in graph.adj[v]:
            if color[neighbor] == GRAY:
                return False  # back edge -> cycle
            if color[neighbor] == WHITE:
                if not dfs(neighbor):
                    return False
        color[v] = BLACK
        result.append(v)
        return True

    for v in graph.vertices:
        if color[v] == WHITE:
            if not dfs(v):
                return None
    return result[::-1]

# Gene regulatory cascade: Growth_Factor -> RAS -> RAF -> MEK -> ERK -> ...
grn = DirectedGraph()
for u, v in [('Growth_Factor', 'RAS'), ('RAS', 'RAF'), ('RAF', 'MEK'),
             ('MEK', 'ERK'), ('ERK', 'Transcription_Factors'),
             ('Transcription_Factors', 'Cell_Cycle_Genes')]:
    grn.add_edge(u, v)

print(topological_sort_dfs(grn))
# ['Growth_Factor', 'RAS', 'RAF', 'MEK', 'ERK', 'Transcription_Factors', 'Cell_Cycle_Genes']

Kahn's Algorithm (BFS) — O(V + E)

Goal: same ordering, computed iteratively instead of recursively (avoids recursion-depth limits on deep DAGs). Approach: repeatedly peel off vertices whose in-degree has dropped to 0. If fewer vertices are emitted than exist in the graph, a cycle remains.

def topological_sort_kahn(graph):
    """Kahn's BFS-based topological sort. Returns ordering, or None if a cycle exists."""
    in_degree = {v: 0 for v in graph.vertices}
    for u in graph.adj:
        for v in graph.adj[u]:
            in_degree[v] += 1

    queue = deque(v for v in graph.vertices if in_degree[v] == 0)
    result = []
    while queue:
        v = queue.popleft()
        result.append(v)
        for neighbor in graph.adj[v]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    return result if len(result) == len(graph.vertices) else None

Swap deque for a heapq (min-heap keyed on vertex label) to get the lexicographically smallest valid ordering. Note: a DAG generally has many valid orderings — neither algorithm's output is unique unless the graph is a single chain.

Critical Path Analysis

Goal: for a weighted DAG (each node has a runtime), find the minimum total makespan achievable with unlimited parallelism, and which nodes sit on the critical (zero-slack) path — e.g. the bottleneck steps of a bioinformatics pipeline. Approach: forward pass in topological order computes each node's earliest start; backward pass computes each node's latest start without delaying the makespan; nodes where earliest == latest have zero slack and are critical.

def critical_path(graph, durations):
    """
    Longest path (in time) through a weighted DAG.

    Parameters
    ----------
    graph     : DirectedGraph
    durations : dict {vertex: runtime}

    Returns
    -------
    (makespan, critical_nodes)
    """
    order = topological_sort_kahn(graph)
    earliest = {v: 0 for v in graph.vertices}
    for v in order:
        finish_v = earliest[v] + durations.get(v, 0)
        for neighbor in graph.adj[v]:
            if finish_v > earliest[neighbor]:
                earliest[neighbor] = finish_v
    makespan = max(earliest[v] + durations.get(v, 0) for v in graph.vertices)

    latest = {v: makespan - durations.get(v, 0) for v in graph.vertices}
    for v in reversed(order):
        for neighbor in graph.adj[v]:
            candidate = latest[neighbor] - durations.get(v, 0)
            if candidate < latest[v]:
                latest[v] = candidate

    critical_nodes = [v for v in order if earliest[v] == latest[v]]
    return makespan, critical_nodes

# RNA-seq pipeline with per-step runtimes (minutes)
pipeline = DirectedGraph()
for u, v in [('FastQC', 'Trimmomatic'), ('Trimmomatic', 'STAR_Alignment'),
             ('STAR_Alignment', 'FeatureCounts'), ('STAR_Alignment', 'BAM_QC'),
             ('FeatureCounts', 'DESeq2'), ('DESeq2', 'Pathway_Analysis'),
             ('BAM_QC', 'MultiQC'), ('FastQC', 'MultiQC')]:
    pipeline.add_edge(u, v)

runtimes = {'FastQC': 5, 'Trimmomatic': 20, 'STAR_Alignment': 45, 'FeatureCounts': 10,
            'BAM_QC': 8, 'DESeq2': 15, 'Pathway_Analysis': 12, 'MultiQC': 3}

makespan, critical = critical_path(pipeline, runtimes)
print(f"Minimum pipeline runtime: {makespan} min")   # 107 min
print(f"Critical path: {' -> '.join(critical)}")      # FastQC -> Trimmomatic -> STAR_Alignment -> FeatureCounts -> DESeq2 -> Pathway_Analysis

Self-Check

def demo():
    """Sanity checks: valid DAG sorts correctly, cycle is detected, critical path is right."""
    g = DirectedGraph()
    for u, v in [('A', 'B'), ('B', 'C'), ('A', 'C')]:
        g.add_edge(u, v)
    assert topological_sort_dfs(g) == ['A', 'B', 'C']
    assert topological_sort_kahn(g) == ['A', 'B', 'C']

    cyclic = DirectedGraph()
    for u, v in [('X', 'Y'), ('Y', 'Z'), ('Z', 'X')]:
        cyclic.add_edge(u, v)
    assert topological_sort_dfs(cyclic) is None
    assert topological_sort_kahn(cyclic) is None

    makespan, critical = critical_path(g, {'A': 1, 'B': 2, 'C': 1})
    assert makespan == 4
    assert critical == ['A', 'B', 'C']
    print("all checks passed")

if __name__ == "__main__":
    demo()

Pitfalls

• DFS topo sort detects cycles via GRAY-to-GRAY back edges; Kahn's detects cycles by comparing len(result) to len(graph.vertices) — both must be checked, don't assume the input is acyclic.
• A DAG can have many valid topological orderings; neither algorithm guarantees a unique or "canonical" result — use a min-heap in Kahn's for the lexicographically smallest one, or backtracking to enumerate all of them (expensive, small graphs only).
• The recursive DFS approach uses O(V) call-stack depth; for very deep/large DAGs (e.g. genome-scale networks) prefer Kahn's iterative version to avoid RecursionError.
• critical_path assumes node durations, not edge weights; if your DAG's cost lives on edges, fold it into a dummy node or adapt the forward/backward pass accordingly.
• Isolated vertices (no edges) are still emitted by both sorts as long as they were registered via add_edge or added to graph.vertices directly — don't forget disconnected pipeline steps.

See Also

• algo-graph-representations — building/choosing adjacency-list vs adjacency-matrix graphs.
• algo-bfs-dfs — the traversal primitives topological sort is built on.
• algo-mst-kruskal-prim — another classic graph algorithm family (undirected, weighted).
• bio-workflow-management-nextflow-pipelines — real pipeline schedulers that resolve DAG dependencies in practice.