Skip to content

graph

A minimal typed directed-acyclic-graph implementation for chain flows.

Deliberately hand-rolled rather than depending on a graph library: chains are small (a handful of nodes), the operations needed are few, and preserving the node key type through generics keeps the call sites fully typed.

ChainGraphCycleError

Bases: ValueError

Raised when an operation requires an acyclic graph but the graph contains a cycle.

Source code in horde_sdk/worker/chaining/graph.py 
13
14
class ChainGraphCycleError(ValueError):
    """Raised when an operation requires an acyclic graph but the graph contains a cycle."""

TypedChainGraph

A small directed graph keyed by hashable node keys, with DAG utilities.

Source code in horde_sdk/worker/chaining/graph.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class TypedChainGraph[NodeKeyT: Hashable]:
    """A small directed graph keyed by hashable node keys, with DAG utilities."""

    def __init__(self) -> None:
        """Initialize an empty graph."""
        self._successors: dict[NodeKeyT, list[NodeKeyT]] = {}
        self._predecessors: dict[NodeKeyT, list[NodeKeyT]] = {}

    def __contains__(self, key: NodeKeyT) -> bool:
        """Return whether the node key is in the graph."""
        return key in self._successors

    def __len__(self) -> int:
        """Return the number of nodes in the graph."""
        return len(self._successors)

    def __iter__(self) -> Iterator[NodeKeyT]:
        """Iterate over the node keys in insertion order."""
        return iter(self._successors)

    def add_node(self, key: NodeKeyT) -> None:
        """Add a node to the graph.

        Args:
            key (NodeKeyT): The node key to add.

        Raises:
            ValueError: If the node already exists.
        """
        if key in self._successors:
            raise ValueError(f"Node {key} already exists in the graph")
        self._successors[key] = []
        self._predecessors[key] = []

    def add_edge(self, source: NodeKeyT, target: NodeKeyT) -> None:
        """Add a directed edge between two existing nodes.

        Args:
            source (NodeKeyT): The edge origin.
            target (NodeKeyT): The edge destination.

        Raises:
            ValueError: If either node is missing, the edge is a self-loop, or the edge already exists.
        """
        if source not in self._successors:
            raise ValueError(f"Source node {source} is not in the graph")
        if target not in self._successors:
            raise ValueError(f"Target node {target} is not in the graph")
        if source == target:
            raise ValueError(f"Self-loop on {source} is not allowed")
        if target in self._successors[source]:
            raise ValueError(f"Edge {source} -> {target} already exists")

        self._successors[source].append(target)
        self._predecessors[target].append(source)

    def successors(self, key: NodeKeyT) -> tuple[NodeKeyT, ...]:
        """Return the direct successors of a node."""
        return tuple(self._successors[key])

    def predecessors(self, key: NodeKeyT) -> tuple[NodeKeyT, ...]:
        """Return the direct predecessors of a node."""
        return tuple(self._predecessors[key])

    def sources(self) -> tuple[NodeKeyT, ...]:
        """Return the nodes with no predecessors."""
        return tuple(key for key, preds in self._predecessors.items() if not preds)

    def sinks(self) -> tuple[NodeKeyT, ...]:
        """Return the nodes with no successors."""
        return tuple(key for key, succs in self._successors.items() if not succs)

    def descendants(self, key: NodeKeyT) -> set[NodeKeyT]:
        """Return all nodes reachable from a node, excluding the node itself."""
        seen: set[NodeKeyT] = set()
        frontier = deque(self._successors[key])
        while frontier:
            current = frontier.popleft()
            if current in seen:
                continue
            seen.add(current)
            frontier.extend(self._successors[current])
        return seen

    def is_connected(self) -> bool:
        """Return whether the graph is weakly connected (or empty)."""
        if not self._successors:
            return True

        start = next(iter(self._successors))
        seen: set[NodeKeyT] = {start}
        frontier = deque([start])
        while frontier:
            current = frontier.popleft()
            for neighbor in (*self._successors[current], *self._predecessors[current]):
                if neighbor not in seen:
                    seen.add(neighbor)
                    frontier.append(neighbor)
        return len(seen) == len(self._successors)

    def topological_order(self) -> tuple[NodeKeyT, ...]:
        """Return the nodes in a topological order (Kahn's algorithm).

        Returns:
            tuple[NodeKeyT, ...]: The nodes such that every edge points from an earlier node to a later one.

        Raises:
            ChainGraphCycleError: If the graph contains a cycle.
        """
        in_degree = {key: len(preds) for key, preds in self._predecessors.items()}
        frontier = deque(key for key, degree in in_degree.items() if degree == 0)
        order: list[NodeKeyT] = []

        while frontier:
            current = frontier.popleft()
            order.append(current)
            for successor in self._successors[current]:
                in_degree[successor] -= 1
                if in_degree[successor] == 0:
                    frontier.append(successor)

        if len(order) != len(self._successors):
            unresolved = sorted(str(key) for key, degree in in_degree.items() if degree > 0)
            raise ChainGraphCycleError(f"Graph contains a cycle involving: {', '.join(unresolved)}")

        return tuple(order)

    def is_directed_acyclic(self) -> bool:
        """Return whether the graph is a DAG."""
        try:
            self.topological_order()
        except ChainGraphCycleError:
            return False
        return True

__init__ 

__init__() -> None

Initialize an empty graph.

Source code in horde_sdk/worker/chaining/graph.py 
20
21
22
23
def __init__(self) -> None:
    """Initialize an empty graph."""
    self._successors: dict[NodeKeyT, list[NodeKeyT]] = {}
    self._predecessors: dict[NodeKeyT, list[NodeKeyT]] = {}

__contains__ 

__contains__(key: NodeKeyT) -> bool

Return whether the node key is in the graph.

Source code in horde_sdk/worker/chaining/graph.py 
25
26
27
def __contains__(self, key: NodeKeyT) -> bool:
    """Return whether the node key is in the graph."""
    return key in self._successors

__len__ 

__len__() -> int

Return the number of nodes in the graph.

Source code in horde_sdk/worker/chaining/graph.py 
29
30
31
def __len__(self) -> int:
    """Return the number of nodes in the graph."""
    return len(self._successors)

__iter__ 

__iter__() -> Iterator[NodeKeyT]

Iterate over the node keys in insertion order.

Source code in horde_sdk/worker/chaining/graph.py 
33
34
35
def __iter__(self) -> Iterator[NodeKeyT]:
    """Iterate over the node keys in insertion order."""
    return iter(self._successors)

add_node 

add_node(key: NodeKeyT) -> None

Add a node to the graph.

Parameters:

  • key (NodeKeyT) –

    The node key to add.

Raises:

  • ValueError –

    If the node already exists.

Source code in horde_sdk/worker/chaining/graph.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
def add_node(self, key: NodeKeyT) -> None:
    """Add a node to the graph.

    Args:
        key (NodeKeyT): The node key to add.

    Raises:
        ValueError: If the node already exists.
    """
    if key in self._successors:
        raise ValueError(f"Node {key} already exists in the graph")
    self._successors[key] = []
    self._predecessors[key] = []

add_edge 

add_edge(source: NodeKeyT, target: NodeKeyT) -> None

Add a directed edge between two existing nodes.

Parameters:

  • source (NodeKeyT) –

    The edge origin.

  • target (NodeKeyT) –

    The edge destination.

Raises:

  • ValueError –

    If either node is missing, the edge is a self-loop, or the edge already exists.

Source code in horde_sdk/worker/chaining/graph.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def add_edge(self, source: NodeKeyT, target: NodeKeyT) -> None:
    """Add a directed edge between two existing nodes.

    Args:
        source (NodeKeyT): The edge origin.
        target (NodeKeyT): The edge destination.

    Raises:
        ValueError: If either node is missing, the edge is a self-loop, or the edge already exists.
    """
    if source not in self._successors:
        raise ValueError(f"Source node {source} is not in the graph")
    if target not in self._successors:
        raise ValueError(f"Target node {target} is not in the graph")
    if source == target:
        raise ValueError(f"Self-loop on {source} is not allowed")
    if target in self._successors[source]:
        raise ValueError(f"Edge {source} -> {target} already exists")

    self._successors[source].append(target)
    self._predecessors[target].append(source)

successors 

successors(key: NodeKeyT) -> tuple[NodeKeyT, ...]

Return the direct successors of a node.

Source code in horde_sdk/worker/chaining/graph.py 
73
74
75
def successors(self, key: NodeKeyT) -> tuple[NodeKeyT, ...]:
    """Return the direct successors of a node."""
    return tuple(self._successors[key])

predecessors 

predecessors(key: NodeKeyT) -> tuple[NodeKeyT, ...]

Return the direct predecessors of a node.

Source code in horde_sdk/worker/chaining/graph.py 
77
78
79
def predecessors(self, key: NodeKeyT) -> tuple[NodeKeyT, ...]:
    """Return the direct predecessors of a node."""
    return tuple(self._predecessors[key])

sources 

sources() -> tuple[NodeKeyT, ...]

Return the nodes with no predecessors.

Source code in horde_sdk/worker/chaining/graph.py 
81
82
83
def sources(self) -> tuple[NodeKeyT, ...]:
    """Return the nodes with no predecessors."""
    return tuple(key for key, preds in self._predecessors.items() if not preds)

sinks 

sinks() -> tuple[NodeKeyT, ...]

Return the nodes with no successors.

Source code in horde_sdk/worker/chaining/graph.py 
85
86
87
def sinks(self) -> tuple[NodeKeyT, ...]:
    """Return the nodes with no successors."""
    return tuple(key for key, succs in self._successors.items() if not succs)

descendants 

descendants(key: NodeKeyT) -> set[NodeKeyT]

Return all nodes reachable from a node, excluding the node itself.

Source code in horde_sdk/worker/chaining/graph.py 
89
90
91
92
93
94
95
96
97
98
99
def descendants(self, key: NodeKeyT) -> set[NodeKeyT]:
    """Return all nodes reachable from a node, excluding the node itself."""
    seen: set[NodeKeyT] = set()
    frontier = deque(self._successors[key])
    while frontier:
        current = frontier.popleft()
        if current in seen:
            continue
        seen.add(current)
        frontier.extend(self._successors[current])
    return seen

is_connected 

is_connected() -> bool

Return whether the graph is weakly connected (or empty).

Source code in horde_sdk/worker/chaining/graph.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def is_connected(self) -> bool:
    """Return whether the graph is weakly connected (or empty)."""
    if not self._successors:
        return True

    start = next(iter(self._successors))
    seen: set[NodeKeyT] = {start}
    frontier = deque([start])
    while frontier:
        current = frontier.popleft()
        for neighbor in (*self._successors[current], *self._predecessors[current]):
            if neighbor not in seen:
                seen.add(neighbor)
                frontier.append(neighbor)
    return len(seen) == len(self._successors)

topological_order 

topological_order() -> tuple[NodeKeyT, ...]

Return the nodes in a topological order (Kahn's algorithm).

Returns:

  • tuple[NodeKeyT, ...] –

    tuple[NodeKeyT, ...]: The nodes such that every edge points from an earlier node to a later one.

Raises:

Source code in horde_sdk/worker/chaining/graph.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def topological_order(self) -> tuple[NodeKeyT, ...]:
    """Return the nodes in a topological order (Kahn's algorithm).

    Returns:
        tuple[NodeKeyT, ...]: The nodes such that every edge points from an earlier node to a later one.

    Raises:
        ChainGraphCycleError: If the graph contains a cycle.
    """
    in_degree = {key: len(preds) for key, preds in self._predecessors.items()}
    frontier = deque(key for key, degree in in_degree.items() if degree == 0)
    order: list[NodeKeyT] = []

    while frontier:
        current = frontier.popleft()
        order.append(current)
        for successor in self._successors[current]:
            in_degree[successor] -= 1
            if in_degree[successor] == 0:
                frontier.append(successor)

    if len(order) != len(self._successors):
        unresolved = sorted(str(key) for key, degree in in_degree.items() if degree > 0)
        raise ChainGraphCycleError(f"Graph contains a cycle involving: {', '.join(unresolved)}")

    return tuple(order)

is_directed_acyclic 

is_directed_acyclic() -> bool

Return whether the graph is a DAG.

Source code in horde_sdk/worker/chaining/graph.py 
144
145
146
147
148
149
150
def is_directed_acyclic(self) -> bool:
    """Return whether the graph is a DAG."""
    try:
        self.topological_order()
    except ChainGraphCycleError:
        return False
    return True