daglib/README.md

daglib

Lightweight directed acyclic graph (DAG) primitives with a mutable mapping-style API.

Install

For local development in this repository:

uv sync --all-groups

Core Concept

DAG behaves like an adjacency map:

• g[u] gives a mutable set-like view of successors of u
• g.reverse[v] gives predecessors of v
• len(g) returns edge count (not node count)

from daglib.dag import DAG

g = DAG[str]()
g.add_edge("build", "test")
g.add_edge("test", "deploy")

assert "test" in g["build"]
assert "build" in g.reverse["test"]
assert len(g) == 2

Adding Edges

Explicit edge API

g = DAG[str]()
g.add_edge("foo", "bar")
g.add_edge("foo", "baz")

Self-loops are rejected:

g = DAG[str]()
g.add_edge("foo", "foo")  # raises ValueError

Mapping and set-like API

g = DAG[str]()

g["foo"] = "bar"              # add single edge
g["foo"] = {"baz", "qux"}     # add multiple edges
g["foo"] += {"zap", "zip"}     # in-place add

assert set(g["foo"]) >= {"bar", "baz", "qux", "zap", "zip"}

Note: assignment is additive in current behavior. It does not clear existing outgoing edges.

Removing Edges and Nodes

g = DAG[str]()
g["foo"] += {"bar", "baz"}

g.remove_edge("foo", "bar")
g["foo"] -= {"baz"}
g.discard_node("foo")

remove_edge supports strict/non-strict behavior:

g = DAG[str]()
g.remove_edge("x", "y", missing_ok=True)   # no error
g.remove_edge("x", "y", missing_ok=False)  # raises KeyError

Iteration and Size

g = DAG[str]()
g.add_edge("foo", "bar")
g.add_edge("bar", "baz")

nodes_with_outgoing = set(g)  # {"foo", "bar"}
edge_count = len(g)           # 2

Reverse View

g = DAG[str]()
g["foo"] = {"baz"}
g["bar"] = {"baz"}

assert set(g.reverse["baz"]) == {"foo", "bar"}

Callbacks

You can hook edge additions/removals done through the mutable set views.

g = DAG[str]()

added = []
removed = []

g.on_add = lambda u, v: added.append((u, v))
g.on_remove = lambda u, v: removed.append((u, v))

g["foo"].add("bar")
g["foo"].discard("bar")

assert ("foo", "bar") in added
assert ("foo", "bar") in removed

Current tested behavior: g["foo"] = {...} does not trigger on_add callbacks.

Topological Sort

g = DAG[str]()
g["build"] = "test"
g["test"] = "deploy"

order = g.topo_sort()
reverse_order = g.topo_sort(reverse=True)

Subgraphs (Transitive Closure from Seeds)

g = DAG[str]()
g["B"] += "C"
g["B"] += "D"
g["D"] += "E"
g["E"] += "F"

sub = g.subgraph("D")
assert set(sub["D"]) == {"E"}
assert set(sub["E"]) == {"F"}

DAGSetView as a Standalone Type

DAGSetView can also be used directly as a mutable set wrapper.

from daglib.set import DAGSetView

s = DAGSetView({"foo", 2, 3})
s.add(4)
s -= {2}
s ^= {3, 5}

assert set(s) == {"foo", 4, 5}

It supports callbacks:

s = DAGSetView()
events = []
events_removed = []

s.on_add = lambda v: events.append(v)
s.on_remove = lambda v: events_removed.append(v)

s |= {1, 2}
s -= {1}

Performance Test Scaffolding

The test suite includes graph generators and a benchmark helper in tests/test_performance.py:

from tests.test_performance import DAGGenerator, benchmark_operation

g = DAGGenerator.chain(10_000)
stats = benchmark_operation(lambda: g.topo_sort(), n_runs=5)
print(stats["avg"])

Run performance tests:

pytest tests/test_performance.py -v -s

Run Tests

pytest