Graph Algorithms Skill

Atomic Responsibility: Execute graph traversal and pathfinding algorithms efficiently.

Graph Representation

from typing import Dict, List, Set, Tuple, Optional
from collections import deque
import heapq

# Adjacency List (most common)
Graph = Dict[int, List[int]]
WeightedGraph = Dict[int, List[Tuple[int, int]]]  # node -> [(neighbor, weight)]

DFS - Depth First Search

def dfs_recursive(graph: Graph, start: int) -> List[int]:
    """
    DFS traversal from start node.

    Time: O(V+E), Space: O(V)

    Args:
        graph: Adjacency list representation
        start: Starting node

    Returns:
        List of nodes in DFS order
    """
    visited: Set[int] = set()
    result: List[int] = []

    def explore(node: int) -> None:
        if node in visited:
            return

        visited.add(node)
        result.append(node)

        for neighbor in graph.get(node, []):
            explore(neighbor)

    explore(start)
    return result


def dfs_iterative(graph: Graph, start: int) -> List[int]:
    """
    Iterative DFS using explicit stack.

    Use when: Avoiding recursion depth limits.
    """
    visited: Set[int] = set()
    result: List[int] = []
    stack = [start]

    while stack:
        node = stack.pop()

        if node in visited:
            continue

        visited.add(node)
        result.append(node)

        # Add neighbors in reverse for correct order
        for neighbor in reversed(graph.get(node, [])):
            if neighbor not in visited:
                stack.append(neighbor)

    return result

BFS - Breadth First Search

def bfs(graph: Graph, start: int) -> List[int]:
    """
    BFS traversal from start node.

    Time: O(V+E), Space: O(V)

    Use for: Shortest path in unweighted graph.
    """
    visited = {start}
    queue = deque([start])
    result: List[int] = []

    while queue:
        node = queue.popleft()
        result.append(node)

        for neighbor in graph.get(node, []):
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append(neighbor)

    return result


def shortest_path_unweighted(graph: Graph, start: int, end: int) -> int:
    """
    Find shortest path length in unweighted graph.

    Time: O(V+E), Space: O(V)

    Returns:
        Shortest distance, or -1 if no path exists
    """
    if start == end:
        return 0

    visited = {start}
    queue = deque([(start, 0)])

    while queue:
        node, distance = queue.popleft()

        for neighbor in graph.get(node, []):
            if neighbor == end:
                return distance + 1

            if neighbor not in visited:
                visited.add(neighbor)
                queue.append((neighbor, distance + 1))

    return -1  # No path found

Dijkstra's Algorithm

def dijkstra(graph: WeightedGraph, start: int) -> Dict[int, int]:
    """
    Single-source shortest paths with non-negative weights.

    Time: O((V+E) log V), Space: O(V)

    Args:
        graph: Weighted adjacency list {node: [(neighbor, weight)]}
        start: Source node

    Returns:
        Dictionary of shortest distances from start

    Raises:
        ValueError: If negative weight detected
    """
    distances: Dict[int, int] = {start: 0}
    pq = [(0, start)]  # (distance, node)

    while pq:
        current_dist, node = heapq.heappop(pq)

        # Skip if we've found a better path
        if current_dist > distances.get(node, float('inf')):
            continue

        for neighbor, weight in graph.get(node, []):
            if weight < 0:
                raise ValueError(f"Negative weight {weight} not allowed in Dijkstra")

            distance = current_dist + weight

            if distance < distances.get(neighbor, float('inf')):
                distances[neighbor] = distance
                heapq.heappush(pq, (distance, neighbor))

    return distances


def dijkstra_with_path(graph: WeightedGraph, start: int, end: int) -> Tuple[int, List[int]]:
    """
    Dijkstra with path reconstruction.

    Returns:
        Tuple of (distance, path) or (inf, []) if no path
    """
    distances: Dict[int, int] = {start: 0}
    predecessors: Dict[int, Optional[int]] = {start: None}
    pq = [(0, start)]

    while pq:
        current_dist, node = heapq.heappop(pq)

        if node == end:
            break

        if current_dist > distances.get(node, float('inf')):
            continue

        for neighbor, weight in graph.get(node, []):
            distance = current_dist + weight

            if distance < distances.get(neighbor, float('inf')):
                distances[neighbor] = distance
                predecessors[neighbor] = node
                heapq.heappush(pq, (distance, neighbor))

    # Reconstruct path
    if end not in distances:
        return float('inf'), []

    path = []
    current = end
    while current is not None:
        path.append(current)
        current = predecessors.get(current)

    return distances[end], path[::-1]

Union-Find (Disjoint Set Union)

class UnionFind:
    """
    Disjoint Set Union with path compression and union by rank.

    Time: O(α(n)) per operation (nearly constant)
    Space: O(n)

    Use for: Connected components, cycle detection, Kruskal's MST
    """

    def __init__(self, n: int):
        """Initialize n disjoint sets."""
        self.parent = list(range(n))
        self.rank = [0] * n
        self.components = n

    def find(self, x: int) -> int:
        """Find root of x with path compression."""
        if self.parent[x] != x:
            self.parent[x] = self.find(self.parent[x])
        return self.parent[x]

    def union(self, x: int, y: int) -> bool:
        """
        Unite sets containing x and y.

        Returns:
            True if union performed, False if already in same set
        """
        px, py = self.find(x), self.find(y)

        if px == py:
            return False

        # Union by rank
        if self.rank[px] < self.rank[py]:
            px, py = py, px

        self.parent[py] = px

        if self.rank[px] == self.rank[py]:
            self.rank[px] += 1

        self.components -= 1
        return True

    def connected(self, x: int, y: int) -> bool:
        """Check if x and y are in the same set."""
        return self.find(x) == self.find(y)

    def get_components(self) -> int:
        """Return number of disjoint sets."""
        return self.components

Topological Sort

def topological_sort_kahn(n: int, edges: List[Tuple[int, int]]) -> List[int]:
    """
    Topological sort using Kahn's algorithm (BFS).

    Time: O(V+E), Space: O(V)

    Args:
        n: Number of nodes (0 to n-1)
        edges: List of (from, to) edges

    Returns:
        Topologically sorted list, or empty if cycle exists
    """
    graph: Graph = {i: [] for i in range(n)}
    in_degree = [0] * n

    for src, dst in edges:
        graph[src].append(dst)
        in_degree[dst] += 1

    queue = deque([i for i in range(n) if in_degree[i] == 0])
    result: List[int] = []

    while queue:
        node = queue.popleft()
        result.append(node)

        for neighbor in graph[node]:
            in_degree[neighbor] -= 1
            if in_degree[neighbor] == 0:
                queue.append(neighbor)

    # Check for cycle
    if len(result) != n:
        return []  # Cycle detected

    return result

Unit Test Template

import pytest

class TestGraphAlgorithms:
    """Unit tests for graph algorithm implementations."""

    @pytest.fixture
    def simple_graph(self):
        return {0: [1, 2], 1: [3], 2: [3], 3: []}

    @pytest.fixture
    def weighted_graph(self):
        return {0: [(1, 4), (2, 1)], 1: [(3, 1)], 2: [(1, 2), (3, 5)], 3: []}

    def test_dfs(self, simple_graph):
        result = dfs_recursive(simple_graph, 0)
        assert 0 in result and 3 in result
        assert result.index(0) < result.index(3)

    def test_bfs_shortest_path(self, simple_graph):
        assert shortest_path_unweighted(simple_graph, 0, 3) == 2

    def test_dijkstra(self, weighted_graph):
        distances = dijkstra(weighted_graph, 0)
        assert distances[3] == 4  # 0->2->1->3 = 1+2+1

    def test_union_find(self):
        uf = UnionFind(5)
        uf.union(0, 1)
        uf.union(2, 3)
        assert uf.connected(0, 1)
        assert not uf.connected(0, 2)
        assert uf.get_components() == 3

    def test_topological_sort(self):
        result = topological_sort_kahn(4, [(0, 1), (0, 2), (1, 3), (2, 3)])
        assert result.index(0) < result.index(1)
        assert result.index(0) < result.index(2)

Troubleshooting

Common Issues

Issue	Cause	Solution
Infinite loop	Not marking visited	Add node to visited before enqueueing
Wrong shortest path	Using DFS instead of BFS	BFS for unweighted, Dijkstra for weighted
Negative weights fail	Dijkstra limitation	Use Bellman-Ford instead
Memory exceeded	Dense graph	Use adjacency list, not matrix

Debug Checklist

□ Visited set initialized correctly?
□ All neighbors processed?
□ Graph representation matches algorithm?
□ Edge cases: empty graph, disconnected?
□ Cycle detection needed?
□ 0-indexed vs 1-indexed nodes?