Graphs and Traversals (BFS and DFS)
Graphs and Traversals (BFS and DFS)
A graph models complex networks of interconnected entities as a set of vertices (nodes) joined by edges. Graphs represent service dependencies, routing topologies, task workflow pipelines, and social graphs. In Python, the boring default is an adjacency list implemented with a dict[T, list[T]]. Traversing graphs with Breadth-First Search (BFS) and Depth-First Search (DFS) provides the basis for shortest path calculation and cycle detection.
Mental model
Adjacency List Representation
An adjacency matrix requires an \(O(V^2)\) grid of memory, which wastes space on sparse networks. An adjacency list stores only actual connections in \(O(V + E)\) space:
network = {
"Depot A": ["Hub B", "Hub C"],
"Hub B": ["Depot A", "Station D"],
"Hub C": ["Depot A", "Station D"],
"Station D": ["Hub B", "Hub C"],
}BFS vs DFS Traversals
- Breadth-First Search (BFS): Uses a FIFO Queue (
collections.deque). It explores outward in concentric ripples (all distance-1 neighbors, then distance-2). Guarantees the shortest path in unweighted graphs. - Depth-First Search (DFS): Uses a LIFO Stack or call-stack recursion. It explores deeply down a single path until hitting a dead end, then backtracks. Ideal for topological ordering and cycle detection.
BFS Waves (Queue): DFS Dive (Stack / Recursion):
(A) (A)
/ | \ /
(B) (C) (D) <- Wave 1 (B)
| |
(E) <- Wave 2 (E) <- Dive deep before (C)
Worked examples
Case 1: BFS for Shortest Hop Delivery Routes
Save as graph_bfs.py. Find the shortest route (fewest transfer hops) between fulfillment hubs in a delivery network.
# graph_bfs.py
from collections import deque
class Graph[T]:
def __init__(self) -> None:
self.adj: dict[T, list[T]] = {}
def add_edge(self, u: T, v: T, bidirectional: bool = True) -> None:
self.adj.setdefault(u, []).append(v)
if bidirectional:
self.adj.setdefault(v, []).append(u)
def shortest_path_bfs(self, start: T, target: T) -> list[T] | None:
if start == target:
return [start]
visited: set[T] = {start}
parent: dict[T, T] = {}
queue: deque[T] = deque([start])
found = False
while queue:
curr = queue.popleft()
if curr == target:
found = True
break
for neighbor in self.adj.get(curr, []):
if neighbor not in visited:
visited.add(neighbor)
parent[neighbor] = curr
queue.append(neighbor)
if not found:
return None
# Reconstruct path
path: list[T] = []
curr_node: T | None = target
while curr_node is not None:
path.append(curr_node)
curr_node = parent.get(curr_node)
path.reverse()
return path
def main() -> None:
network: Graph[str] = Graph()
network.add_edge("Central-Depot", "North-Hub")
network.add_edge("Central-Depot", "South-Hub")
network.add_edge("North-Hub", "Station-East")
network.add_edge("South-Hub", "Station-West")
network.add_edge("Station-East", "Terminal-Z")
network.add_edge("Station-West", "Terminal-Z")
start = "Central-Depot"
target = "Terminal-Z"
route = network.shortest_path_bfs(start, target)
assert route is not None
print(f"BFS Shortest Route ({len(route) - 1} hops):")
print(" -> ".join(route))
if __name__ == "__main__":
main()Run:
uv run python graph_bfs.pyOutput:
BFS Shortest Route (3 hops):
Central-Depot -> North-Hub -> Station-East -> Terminal-Z
Case 2: DFS for Workflow Cycle and Deadlock Detection
Save as graph_dfs_cycle.py. Detect whether a task pipeline contains circular dependencies before dispatching batch jobs.
# graph_dfs_cycle.py
class TaskPipeline:
def __init__(self) -> None:
self.adj: dict[str, list[str]] = {}
def add_dependency(self, task: str, depends_on: str) -> None:
# task -> depends_on
self.adj.setdefault(task, []).append(depends_on)
def has_cycle(self) -> bool:
# 3-color DFS:
# 0: Unvisited, 1: Currently Visiting (in call stack), 2: Fully Visited
state: dict[str, int] = {}
def _dfs(node: str) -> bool:
state[node] = 1 # Visiting
for neighbor in self.adj.get(node, []):
neighbor_state = state.get(neighbor, 0)
if neighbor_state == 1:
# Found back-edge to ancestor in current call stack!
return True
if neighbor_state == 0:
if _dfs(neighbor):
return True
state[node] = 2 # Fully visited
return False
all_nodes = set(self.adj.keys())
for u in self.adj.values():
all_nodes.update(u)
for node in all_nodes:
if state.get(node, 0) == 0:
if _dfs(node):
return True
return False
def main() -> None:
pipeline = TaskPipeline()
pipeline.add_dependency("Generate-Invoice", "Verify-Items")
pipeline.add_dependency("Verify-Items", "Calculate-Tax")
pipeline.add_dependency("Calculate-Tax", "Read-Cart")
print(f"Valid linear pipeline has cycle? {pipeline.has_cycle()}")
print("Injecting circular dependency: Read-Cart -> Generate-Invoice")
pipeline.add_dependency("Read-Cart", "Generate-Invoice")
print(f"Pipeline now has cycle? {pipeline.has_cycle()}")
if __name__ == "__main__":
main()Run:
uv run python graph_dfs_cycle.pyOutput:
Valid linear pipeline has cycle? False
Injecting circular dependency: Read-Cart -> Generate-Invoice
Pipeline now has cycle? True
The trap
The infinite cycle recursion trap: If a graph contains cycles and your traversal omits a visited set, recursive DFS will repeatedly re-visit earlier vertices until throwing RecursionError: maximum recursion depth exceeded. Always maintain an explicit visited set or 3-color status map.
The boring rule
- Represent graphs using a
dict[T, list[T]]adjacency list. - For unweighted shortest path queries, always use BFS with
collections.deque. - For cycle detection, dependency scheduling, or topological sorting, use DFS with recursion or an explicit stack.
- When reconstructing routes, record a
parentmapping during traversal rather than copying list instances into the queue on every hop.
Try this
- Implement topological sorting (
topological_sort() -> list[str]) using DFS post-order traversal to produce an ordered build plan. - Write a function
connected_components(graph) -> list[set[str]]that partitions an unpartitioned routing graph into isolated subnet clusters.