Modern API

API reference for the typed-handles and uniqueness modules — NodeRef, @dagron.flow, Effect, dagron.reactive, dagron.contentcache, and dagron.trace. For a walkthrough of how they compose, see the Typed Handles & Reactive Engine guide.

dagron.NodeRef

class NodeRef:
    name: str
    epoch: int

Stable handle returned by DAG.add_node(). Survives unrelated graph mutations; invalidated only when the underlying node is removed (or remove-and-readded with the same name, which yields a fresh epoch).

dag.node_ref(name: str) -> NodeRef | None

Look up the current ref for a name without mutating the DAG.

Every public method that previously took name: str now takes NodeArg = str | NodeRef: add_edge, remove_node, has_node, has_edge, get_payload, set_payload, predecessors, successors, ancestors, descendants, subgraph, subgraph_by_depth, collapse, dominator_tree, all_paths, shortest_path, longest_path, dirty_set, change_provenance, is_ancestor, and the ReachabilityIndex query methods.

Stale refs raise dagron.StaleNodeRefError.

dagron.flow

@dagron.task
def fn(...) -> T: ...

@dagron.task(effect=Effect.NETWORK)
def fn(...) -> T: ...

@dagron.flow
def pipeline(...) -> FlowFuture[T] | None: ...

batch() semantics aren't part of dagron.flow — they live in dagron.reactive.

dagron.Effect

class Effect(Enum):
    PURE              = "pure"
    READ              = "read"
    WRITE             = "write"
    NETWORK           = "network"
    NONDETERMINISTIC  = "nondeterministic"

    is_cacheable: bool
    is_deterministic: bool
    is_isolated: bool

def effects_of(dag: DAG) -> dict[str, Effect]

Read every node's effect tag from a DAG built by @dagron.flow. Returns Effect.PURE for nodes without a tag.

DAGExecutor(enforce_effect_isolation=True) reads these tags and runs NONDETERMINISTIC nodes through a shared lock, so they don't overlap.

dagron.reactive

import dagron.reactive as dr

s = dr.signal(value)              # → Signal[T]
c = dr.computed(lambda: ...)      # → Computed[T]
w = dr.watch(lambda: ...)         # → Watcher (also fires once now)
with dr.batch(): ...              # glitch-free coalesced updates

Track via thread-local; reads inside a Computed body or Watcher body register the source as a dep. Observers are held by weakref.WeakSet so dropped derived nodes don't leak.

dagron.contentcache

from dagron.contentcache import ContentCache, default_cache_dir

cache = ContentCache(cache_dir=None, hasher=None)
cache.compute_or_cached(fn, args=(), kwargs=None, effect=None) -> tuple[Any, bool]
cache.get(fingerprint: bytes) -> tuple[Any, bool]
cache.put(fingerprint: bytes, value: Any) -> None
cache.has(fingerprint: bytes) -> bool
cache.delete(fingerprint: bytes) -> None
cache.clear() -> None
cache.hash(value: Any) -> bytes      # delegates to the configured Hasher

compute_or_cached is effect-aware: WRITE / NETWORK / NONDETERMINISTIC skip the cache entirely; PURE and READ go through it.

Storage layout: <cache_dir>/<aa>/<bb>/<rest>.cache where the fingerprint hex is <aa><bb><rest>. POSIX rename(2) makes writes atomic. The filesystem itself is the index — independent processes share intermediates with no coordination.

dagron.trace

from dagron.trace import TraceWriter, TraceReader, TraceRecord, ReplayedNode, replay

writer = TraceWriter(path, cas=None)
writer.record(name, *, value=None, effect=None, duration_ns=0,
              error=None, metadata=None, timestamp=None) -> TraceRecord
writer.flush()
writer.close()                # also via context manager

reader = TraceReader(path, cas=None)
reader.records() -> Iterator[TraceRecord]
reader.records_until(t, *, inclusive=True) -> Iterator[TraceRecord]
reader.timeline() -> list[tuple[float, str]]
reader.fetch(rec) -> Any      # resolves payload via the CAS

replay(source, *, at=None, cas=None) -> dict[str, ReplayedNode]

ReplayedNode carries name, timestamp, value, effect, replayable, duration_ns, error, and a derived has_value property. replayable mirrors effect.is_deterministic: pure / READ nodes can be reproduced; impure nodes' values are what that run produced, not what a fresh run would produce.

Logs are append-only JSONL. Payloads live in the bound ContentCache, deduplicated across runs that produced the same value.

dagron.stubgen

from dagron.stubgen import generate_stub

generate_stub(
    dag,
    *,
    type_hints: dict[str, type | str] | None = None,
    tasks: dict[str, Callable] | None = None,
    name: str = "TypedExecutionResult",
) -> str

Emits .pyi-formatted source declaring a class with Literal["nodename"] overloads typed by inferred (or explicitly provided) return types. Drop the result into a stub file alongside your code so even string-keyed result["nodename"] lookups become statically typed.