Graphs inside objects
Dependencies, routes and social links as adjacency behind an API.
01The intuition
The problem it solves
A build tool must run tasks in an order that respects "compile before test, test before package". The first version stores each task's prerequisites as strings and runs a task when it notices all of them are done, looping until nothing changes.
tasks = {"package": ["test"], "test": ["compile"], "compile": [], "docs": ["compile"]}
def run_all() -> list[str]:
done: list[str] = []
while len(done) < len(tasks):
for name, prereqs in tasks.items():
if name not in done and all(p in done for p in prereqs):
done.append(name)
return done
Three faults. A cycle, such as test needing package, makes the loop spin forever. Each pass rescans every task, so n tasks cost about n² checks. And the relationships live in a bare dictionary with no place to attach a rule such as "a task cannot depend on itself". The shape of the data is a graph, and it deserves an object.
Build it one piece at a time
Step one: wrap the adjacency in a class with the two operations the problem needs, add an edge and list dependents, and validate on the way in. A dictionary from node to the set of nodes it points to is the adjacency list; it is the default representation for sparse graphs.
class Graph:
def __init__(self) -> None:
self.out: dict[str, set[str]] = {} # node -> nodes it points to
def add_node(self, n: str) -> None:
self.out.setdefault(n, set())
def add_edge(self, a: str, b: str) -> None:
if a == b:
raise ValueError("a task cannot depend on itself")
self.add_node(a); self.add_node(b)
self.out[a].add(b) # a must finish before b
def successors(self, n: str) -> set[str]:
return self.out[n]
Step two: an order that respects every edge is a topological sort. Count incoming edges per node, start with the nodes that have none, and each time a node is emitted, reduce the count of its successors; a successor whose count reaches zero is ready. Every edge is looked at once.
from collections import deque
class Graph:
# add_node, add_edge, successors as above
def topo_order(self) -> list[str]:
indeg = {n: 0 for n in self.out}
for n in self.out:
for m in self.out[n]:
indeg[m] += 1
ready = deque(n for n, d in indeg.items() if d == 0)
order: list[str] = []
while ready:
n = ready.popleft()
order.append(n)
for m in self.out[n]:
indeg[m] -= 1
if indeg[m] == 0:
ready.append(m)
return order
Step three: detect the cycle instead of spinning. If the order has fewer nodes than the graph, the leftovers form a cycle; report them. The same traversal, with a visited set, answers reachability questions such as "does changing compile affect docs".
class Graph(Graph): # continued
def check_acyclic(self) -> None:
order = self.topo_order()
if len(order) != len(self.out):
stuck = sorted(set(self.out) - set(order))
raise ValueError(f"cycle among {stuck}")
def reachable_from(self, start: str) -> set[str]:
seen: set[str] = set()
stack = [start]
while stack:
n = stack.pop()
for m in self.out[n]:
if m not in seen:
seen.add(m); stack.append(m)
return seen
Kahn's algorithm on the build graph:
| Step | ready queue | emitted | indegree after |
|---|---|---|---|
| start | [compile] | test 1, docs 1, package 1 | |
| 1 | [test, docs] | compile | test 0, docs 0, package 1 |
| 2 | [docs, package] | test | package 0 |
| 3 | [package] | docs | |
| 4 | [] | package |
Why it works
The invariant of the topological sort is that a node enters the ready queue only when every node pointing to it has already been emitted, so no edge is ever violated in the output. Because each edge decrements exactly one counter once, the whole thing costs the number of nodes plus the number of edges. The cycle check follows from the same invariant: a node on a cycle never reaches indegree zero.
Where it breaks: the adjacency list is one-directional. If the problem also asks "who depends on me", keep a second dictionary of incoming edges and update both in add_edge, or the two drift apart. Deep recursive traversals also hit Python's recursion limit on long chains, which is why the examples use explicit stacks and queues.
naive loop: up to n passes × n tasks = n². Kahn: n + e, with e the number of edges
The trade-off you must say out loud
An adjacency list is the right default for sparse graphs; an adjacency matrix wins only when the graph is dense or edge lookups must be O(1). Hiding the representation behind add_edge and successors means that choice can change later. Say also what you are not building: shortest paths and weights unless the problem asks. The follow-up that usually comes next is dynamic change, an edge added while a traversal runs, which is where snapshots or a version counter enter.
Where it shows up in the 100 Google problems
- Task scheduler with dependencies: the build graph above, with retries and priorities layered on the ready queue.
- Spreadsheet with formulas: cells reference cells; recalculation is a topological order and a cycle is an error shown to the user.
- Variable store with placeholder substitution: %A% referring to %B% is an edge; cycle detection stops infinite expansion.
- Splitwise: who owes whom is a weighted directed graph; simplifying debts reduces its edges.
- Employee hierarchy: a tree is a graph with one parent per node; is_manager is reachability.