A Production Graph Adjacency List With Node Metadata
The bare adjacency list in the official catalog is a starting point. In production I need node metadata, safe edge deletion, and a traversal helper that does not break when a node is removed mid-walk.
By @kavyachakraborty
December 20, 2025
·
Updated May 20, 2026
706 views
5
4.2 (10)
from __future__ import annotations
from collections import defaultdict
from typing import Any, Iterable
# A real graph almost always carries metadata: a node has a label, a created_at,
# maybe a status. An edge has a relationship type. Plain { node: [neighbors] }
# loses both. Two parallel structures keep them.
class Graph:
def __init__(self) -> None:
self.nodes: dict[str, dict] = {}
self.edges: dict[str, dict[str, str]] = defaultdict(dict) # src -> {dst: rel}
def add_node(self, node_id: str, **data: Any) -> None:
self.nodes[node_id] = data
def add_edge(self, src: str, dst: str, rel: str = 'links_to') -> None:
if src not in self.nodes or dst not in self.nodes:
raise KeyError(f'unknown node: {src!r} or {dst!r}')
self.edges[src][dst] = rel
def neighbors(self, node_id: str) -> Iterable[tuple[str, str]]:
return list(self.edges.get(node_id, {}).items())
g = Graph()
g.add_node('article:1', title='How I sized my JVM heap', author='alice')
g.add_node('user:alice', name='Alice', joined='2024-01-15')
g.add_node('article:2', title='Tracing in 30 lines', author='alice')
g.add_edge('user:alice', 'article:1', 'authored')
g.add_edge('user:alice', 'article:2', 'authored')
g.add_edge('article:1', 'article:2', 'see_also')
print(g.nodes['article:1'])
for dst, rel in g.neighbors('user:alice'):
print(f'user:alice --{rel}--> {dst} ({g.nodes[dst]["title"]})')Two stores, one purpose: nodes holds the metadata dict per id, and edges holds a src -> {dst: rel} map so each edge has a typed relationship. The cost of the metadata dict is one extra hash lookup per node access, which is negligible compared to having to denormalize node data into every edge. The add_edge checks both endpoints; that single line has caught dozens of typos in production. The neighbors method returns a list (not the underlying dict view) so the caller can safely mutate the graph during iteration, which the next accordion needs.
from __future__ import annotations
from collections import defaultdict
from typing import Any, Iterable
class Graph:
def __init__(self) -> None:
self.nodes: dict[str, dict] = {}
self.edges: dict[str, dict[str, str]] = defaultdict(dict)
def add_node(self, node_id: str, **data: Any) -> None:
self.nodes[node_id] = data
def add_edge(self, src: str, dst: str, rel: str = 'links_to') -> None:
if src not in self.nodes or dst not in self.nodes:
raise KeyError(f'unknown node: {src!r} or {dst!r}')
self.edges[src][dst] = rel
def remove_node(self, node_id: str) -> None:
# 1. Drop the node itself.
self.nodes.pop(node_id, None)
# 2. Drop outgoing edges from this node.
self.edges.pop(node_id, None)
# 3. Drop INCOMING edges from every other node.
# This is the half people forget; it leaves dangling references and
# the next traversal blows up with a KeyError on g.nodes[dst].
for src, dsts in list(self.edges.items()):
if node_id in dsts:
del dsts[node_id]
if not dsts:
del self.edges[src]
def neighbors(self, node_id: str) -> Iterable[tuple[str, str]]:
return list(self.edges.get(node_id, {}).items())
g = Graph()
for n in ('a', 'b', 'c', 'd'):
g.add_node(n, label=n.upper())
g.add_edge('a', 'b'); g.add_edge('b', 'c'); g.add_edge('c', 'a'); g.add_edge('d', 'a')
print('before:', dict(g.edges))
g.remove_node('a')
print('after :', dict(g.edges))The bug I have shipped most often in graph code is forgetting step 3, the incoming-edge sweep. Without it, removing node a leaves c -> a and d -> a references pointing at a node that no longer exists in self.nodes, and the next traversal raises KeyError. Because incoming edges are not indexed, the sweep is O(E), which is why a real production graph keeps a reverse index when deletions are common. The if not dsts: del self.edges[src] cleanup is cosmetic but keeps the structure honest: empty entries are surprising during debugging.
from __future__ import annotations
from collections import defaultdict, deque
from typing import Any, Iterator
class Graph:
def __init__(self) -> None:
self.nodes: dict[str, dict] = {}
self.edges: dict[str, dict[str, str]] = defaultdict(dict)
def add_node(self, node_id: str, **data: Any) -> None:
self.nodes[node_id] = data
def add_edge(self, src: str, dst: str, rel: str = 'links_to') -> None:
self.edges[src][dst] = rel
def bfs(self, start: str) -> Iterator[str]:
if start not in self.nodes:
return
seen = {start}
q = deque([start])
while q:
node = q.popleft()
yield node
# Snapshot the neighbors at visit time. If the caller mutates the
# graph (e.g. deletes nodes) during iteration we still walk the
# set we committed to, instead of crashing or skipping nodes.
neighbors = list(self.edges.get(node, {}).keys())
for nb in neighbors:
if nb in self.nodes and nb not in seen:
seen.add(nb)
q.append(nb)
g = Graph()
for n in ('root', 'a', 'b', 'c', 'd'):
g.add_node(n, label=n)
g.add_edge('root', 'a'); g.add_edge('root', 'b')
g.add_edge('a', 'c'); g.add_edge('b', 'd')
for node in g.bfs('root'):
print('visit', node, ' payload:', g.nodes[node])Two design choices make BFS safe under mutation: snapshotting neighbors into a list at visit time, and re-checking nb in self.nodes before enqueueing. The snapshot prevents RuntimeError: dictionary changed size during iteration when the caller deletes an edge as a side effect of visiting a node. The membership re-check makes the traversal forgiving when a downstream node is removed mid-walk; we just skip it rather than crashing. I use this exact pattern in a content-graph crawler where visiting an article can delete it (because it is now flagged spam); the traversal needs to stay correct even though the underlying graph is moving.