Traversal and History¶
This notebook covers the lightweight algorithmic surface that already exists and the built-in mutation log.
In [1]:
Copied!
import sys
from pathlib import Path
repo_root = Path.cwd()
if not (repo_root / 'annnet').exists():
for parent in repo_root.parents:
if (parent / 'annnet').exists():
repo_root = parent
break
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
import annnet as an
import sys
from pathlib import Path
repo_root = Path.cwd()
if not (repo_root / 'annnet').exists():
for parent in repo_root.parents:
if (parent / 'annnet').exists():
repo_root = parent
break
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
import annnet as an
In [2]:
Copied!
G = an.AnnNet(directed=None)
G.add_vertices_bulk(['Signal', 'Kinase', 'TF', 'Target'])
G.add_edge('Signal', 'Kinase', edge_id='e_signal', directed=True)
G.add_edge('Kinase', 'TF', edge_id='e_activation', directed=True)
G.add_edge('TF', 'Target', edge_id='e_regulation', directed=True)
G.add_edge(src=['Signal', 'TF', 'Target'], edge_id='h_context', directed=False)
G = an.AnnNet(directed=None)
G.add_vertices_bulk(['Signal', 'Kinase', 'TF', 'Target'])
G.add_edge('Signal', 'Kinase', edge_id='e_signal', directed=True)
G.add_edge('Kinase', 'TF', edge_id='e_activation', directed=True)
G.add_edge('TF', 'Target', edge_id='e_regulation', directed=True)
G.add_edge(src=['Signal', 'TF', 'Target'], edge_id='h_context', directed=False)
Out[2]:
'h_context'
Local traversal¶
Traversal works directly against the AnnNet structure; it does not require an external backend.
In [3]:
Copied!
print('neighbors(Kinase):', sorted(G.neighbors('Kinase')))
print('out_neighbors(Kinase):', sorted(G.out_neighbors('Kinase')))
print('in_neighbors(Target):', sorted(G.in_neighbors('Target')))
print('successors(TF):', sorted(G.successors('TF')))
print('predecessors(TF):', sorted(G.predecessors('TF')))
print('neighbors(Kinase):', sorted(G.neighbors('Kinase')))
print('out_neighbors(Kinase):', sorted(G.out_neighbors('Kinase')))
print('in_neighbors(Target):', sorted(G.in_neighbors('Target')))
print('successors(TF):', sorted(G.successors('TF')))
print('predecessors(TF):', sorted(G.predecessors('TF')))
neighbors(Kinase): ['TF'] out_neighbors(Kinase): ['TF'] in_neighbors(Target): ['Signal', 'TF'] successors(TF): ['Signal', 'Target'] predecessors(TF): ['Kinase', 'Signal', 'Target']
History¶
Mutating methods are wrapped automatically. The history log records operation name, version, timestamps, arguments, and result.
In [4]:
Copied!
events = G.history()
print('history length:', len(events))
print('last event:', events[-1])
G.history(as_df=True)
events = G.history()
print('history length:', len(events))
print('last event:', events[-1])
G.history(as_df=True)
history length: 4
last event: {'version': 4, 'ts_utc': '2026-04-14T10:50:36.580655Z', 'mono_ns': 8331811, 'op': 'add_edge', 'src': ['Signal', 'TF', 'Target'], 'tgt': None, 'weight': 1.0, 'edge_id': 'h_context', 'directed': False, 'parallel': 'update', 'slice': None, 'as_entity': False, 'propagate': 'none', 'flexible': None, 'attrs': {}, 'result': 'h_context'}
Out[4]:
| version | ts_utc | mono_ns | op | src | tgt | weight | edge_id | directed | parallel | slice | as_entity | propagate | flexible | attrs | result | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 1 | 2026-04-14T10:50:36.580359Z | 8052673 | add_edge | Signal | Kinase | 1.0 | e_signal | True | update | None | False | none | None | {} | e_signal |
| 1 | 2 | 2026-04-14T10:50:36.580492Z | 8171530 | add_edge | Kinase | TF | 1.0 | e_activation | True | update | None | False | none | None | {} | e_activation |
| 2 | 3 | 2026-04-14T10:50:36.580578Z | 8255385 | add_edge | TF | Target | 1.0 | e_regulation | True | update | None | False | none | None | {} | e_regulation |
| 3 | 4 | 2026-04-14T10:50:36.580655Z | 8331811 | add_edge | [Signal, TF, Target] | None | 1.0 | h_context | False | update | None | False | none | None | {} | h_context |
Snapshots and diffs¶
Snapshots are lightweight structural summaries that make it easy to compare graph states.
In [5]:
Copied!
before = G.snapshot('before_extra_edge')
G.add_edge('Signal', 'Target', edge_id='e_shortcut', directed=True)
after = G.snapshot('after_extra_edge')
diff = G.diff('before_extra_edge', 'after_extra_edge')
print(diff)
print(diff.to_dict())
before = G.snapshot('before_extra_edge')
G.add_edge('Signal', 'Target', edge_id='e_shortcut', directed=True)
after = G.snapshot('after_extra_edge')
diff = G.diff('before_extra_edge', 'after_extra_edge')
print(diff)
print(diff.to_dict())
Diff: before_extra_edge - after_extra_edge
Vertices: +0 added, 0 removed
Edges: +1 added, 0 removed
slices: +0 added, 0 removed
{'snapshot_a': 'before_extra_edge', 'snapshot_b': 'after_extra_edge', 'vertices_added': [], 'vertices_removed': [], 'edges_added': ['e_shortcut'], 'edges_removed': [], 'slices_added': [], 'slices_removed': []}