API Reference

Technical documentation for the Ancestree lineage system. The LineageStore is the public entry point and is accessible directly from the top-level ancestree package — searching, lineage traversal, and visualisation all happen through its methods.

Core Orchestration

The LineageStore is the prime entry point for managing your pipeline.

ancestree.LineageStore

Orchestrates the lineage and interactions of a data pipeline.

The LineageStore manages the physical storage, rule enforcement, and hierarchical relationships between different steps in a data pipeline. The rules need only be specified once as configurations persist. The LineageStore does not need to exist in memory. It can be recreated any time it is required. Provides advanced searching capabilities across the node network.

Source code in src/ancestree/core.py

class LineageStore:
    """
    Orchestrates the lineage and interactions of a data pipeline.

    The LineageStore manages the physical storage, rule enforcement, and hierarchical relationships between different steps in a data pipeline.
    The rules need only be specified once as configurations persist. The LineageStore does not need to exist in memory. It can be recreated any time it is required.
    Provides advanced searching capabilities across the node network.
    """

    def __init__(
        self,
        root: Union[Path, str],
        rules: Optional[Dict[str, Any]] = None,
        gen_triggers: Optional[List[str]] = None,
        dedupe: bool = True,
        chunk: bool = True,
    ):
        """
        Initialises the LineageStore, ensures its directory exists, and loads or creates the ruleset configuration.

        On creation the LineageStore saves a .lineage_config.json file. On subsequent re-creation, the store reads from this file. There is no need to resupply rules or gen_triggers at any point after initial creation even if the store no longer exists in memory. The rules and gen_triggers cannot be changed after initial creation.

        Args:
            root (Union[Path, str]): Root directory for data pipeline. This is where the nodes sit.
            rules (Dict, optional): A mapping defining the allowed transitions. Defaults to None.
            gen_triggers (List, optional): Step types that mark a new generation. When a node of this type is created, its generation number increments by one relative to its parent. Defaults to None.
            dedupe (bool, optional): When True, a node that is content-identical to one already in the store is not created a second time: `create_node` reuses the existing node instead, and the variable yielded by the `with` block points at it. Two nodes are content-identical when they share the same step_type, the same parent, the same user metadata, and byte-identical artifacts; volatile fields (node_id, timestamp, duration, size) and provenance (user, platform, git state) are ignored. A candidate match is byte-verified before reuse. This is a behaviour of the store instance and is not persisted in the config — pass it each time you open a store you want to deduplicate. Defaults to False.
            chunk (bool, optional): When True, artifacts are deduplicated at sub-file granularity. As each node is persisted, its files are split into content-defined chunks stored once in a shared pool (`<root>/.chunks`); near-identical artifacts across nodes then cost only their differing chunks. This is transparent: you write files normally inside `create_node`, and reading them back (`node / "file"`, `node.artifacts()`, the web graph) reassembles them on demand. Space is reclaimed by calling `compact()`. Like `dedupe`, this is a per-instance behaviour and is not persisted in the config. Defaults to False.

        Examples:
            >>> rules = {"clean": ["ingest"], "model": ["clean"]}
            >>> store = LineageStore("my_project", rules=rules, gen_triggers=["ingest"])
        """
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)

        self.config_path = self.root / ".lineage_config.json"
        config = self._do_config(rules, gen_triggers)
        self.rules = config["rules"]
        self.triggers = config["triggers"]
        self.dedupe = dedupe
        self.chunk = chunk
        self.database = lineage_database(self.root)

        # Deferred packing: artifacts are written as normal files (native speed)
        # and chunked into the pool by a background worker, off the write path.
        # The worker only ever ADDS (chunks + manifest); loose files are removed
        # only by flush()/close, so an interrupt never loses data.
        self._pack_queue: "queue.Queue[str]" = queue.Queue()
        self._pack_worker: Optional[threading.Thread] = None
        self._pack_lock = threading.Lock()
        self._enqueued: Set[str] = set()  # node_ids queued for chunking
        self._reclaim_pending: Set[str] = set()  # chunked, awaiting loose reclaim
        self._flush_registered = False
        self._scan_done = threading.Event()  # worker finished its straggler scan
        # Track this store so the after-fork handler can re-arm its packer in a
        # forked child (a thread does not survive fork — see _reset_workers_after_fork).
        _live_stores.add(self)

    def __enter__(self) -> "LineageStore":
        return self

    def __exit__(self, *exc: Any) -> None:
        # Finish deferred packing and reclaim loose files (the only point loose
        # files are removed — quiescent, so no reader can be mid-read), then wipe
        # the session read cache. flush() is also run at interpreter exit, so
        # non-context-manager use converges too.
        self.flush()
        self.clear_cache()

    def clear_cache(self) -> None:
        """
        Discards the store's session read cache (`<root>/.cache`).

        Reading a packed artifact reassembles its bytes into this cache rather
        than back into the node directory, so the store never needs `compact()`
        to reclaim what reads would otherwise leak. The cache is pure derived
        data — anything in it is regenerated from the chunk pool on the next
        read — so clearing it only frees disk, never loses anything. It is also
        cleared automatically when the process exits or a `with` block closes;
        call this to reclaim the space sooner.
        """
        drop_read_cache(self.root)

    def _do_config(
        self,
        supplied_rules: Optional[Dict[str, Any]],
        supplied_triggers: Optional[List[str]],
    ) -> Dict[str, Any]:
        is_new = not self.config_path.exists()

        if is_new:
            tmp = self.root / f".lineage_config.{uuid.uuid4().hex}.tmp"
            try:
                tmp.write_text(
                    json.dumps(
                        {"rules": supplied_rules, "triggers": supplied_triggers},
                        indent=2,
                    )
                )
                tmp.replace(self.config_path)
            finally:
                tmp.unlink(missing_ok=True)

        config = json.loads(self.config_path.read_text())

        if not is_new:
            stored_rules = config.get("rules") or {}
            stored_triggers = config.get("triggers") or []
            if supplied_rules and supplied_rules != stored_rules:
                warnings.warn(
                    "Supplied rules differ from the stored configuration and have been ignored. "
                    "Rules cannot be changed after a store is created.",
                    UserWarning,
                    stacklevel=3,
                )
            if supplied_triggers and supplied_triggers != stored_triggers:
                warnings.warn(
                    "Supplied gen_triggers differ from the stored configuration and have been ignored. "
                    "gen_triggers cannot be changed after a store is created.",
                    UserWarning,
                    stacklevel=3,
                )

        return {
            "rules": config.get("rules") or {},
            "triggers": config.get("triggers") or [],
        }

    # ---------------------------------------------------------------------------
    # Node creation
    # ---------------------------------------------------------------------------

    @contextmanager
    def create_node(
        self,
        step_type: str,
        parent: Union["Node", str, List[Union["Node", str]], None] = None,
    ) -> Iterator["Node"]:
        """Creates a new node while enforcing lineage rules.

        The node only materialises on disk once the user writes an artifact or
        adds metadata; an untouched node is discarded with a warning. If the
        user's code raises after writing, the partial work is persisted and the
        node's 'healthy' metadata flag is set to False (True on clean completion).

        If the store was opened with `dedupe=True` and the block completes
        cleanly, a node that is content-identical to one already in the store
        is not created again: the existing node is reused and the `as node`
        variable is rebound onto it. See `LineageStore.__init__` for what counts
        as content-identical.

        Args:
            step_type (str): The type of pipeline step being performed.
            parent (Node | str | list, optional): The parent — a Node or node_id, or a **list** of them for a step that joins several inputs (a merge/join, making the lineage a DAG). Every parent must already exist in this store and individually satisfy the rules; the node's generation is one past its deepest parent. Defaults to None (a root). Examples: `parent=clean`, `parent=[features_a, features_b]`.

        Raises:
            ValueError: If the step type transition is not permitted according to the store rules.

        Yields:
            Node: The new node, ready to receive artifacts and metadata.

        Examples:
            >>> with store.create_node(step_type="clean", parent=ingest_node) as node:
            ...     df.to_csv(node / "cleaned.csv")
            ...     node.add_meta("rows", len(df))
        """

        self._validate_step_type(step_type)
        parents = self._resolve_parents(parent)

        # Rule check: every parent must individually be a legal predecessor.
        allowed = self.rules.get(step_type)
        if allowed is not None:
            parent_types = [p.step_type for p in parents] or [None]
            for ptype in parent_types:
                if ptype not in allowed:
                    raise ValueError(
                        f"Invalid transition: {ptype} -> {step_type}. "
                        f"Allowed parents: {allowed}."
                    )

        # A node sits one generation below its deepest input (when it triggers a
        # new generation), otherwise at that depth.
        base_gen = max((p.generation for p in parents), default=0)
        current_gen = (
            base_gen + 1 if (parents and step_type in self.triggers) else base_gen
        )

        node_id = uuid.uuid4().hex[:8]
        while node_id in self.database.cache or (self.root / node_id).exists():
            node_id = uuid.uuid4().hex[:8]
        node_path = self.root / node_id

        parent_id = [p.node_id for p in parents]
        new_node = Node._create(
            node_path, node_id, current_gen, parent_id, step_type=step_type
        )

        start = time.monotonic()
        try:
            yield new_node
        except BaseException:
            # Keep partial work: anything written before the failure persists,
            # flagged as unhealthy. An untouched node leaves no trace.
            if not self._persist_if_touched(
                new_node, healthy=False, duration=time.monotonic() - start
            ):
                shutil.rmtree(new_node.path, ignore_errors=True)
            raise

        if not self._persist_if_touched(
            new_node, healthy=True, duration=time.monotonic() - start
        ):
            shutil.rmtree(new_node.path, ignore_errors=True)

            warnings.warn(
                f"Node '{new_node.node_id}' (step_type='{step_type}') was discarded: "
                "no artifacts were written and no metadata was added. "
                "Write at least one file or call node.add_meta() to persist the node.",
                UserWarning,
                # 1=here, 2=contextlib.__exit__ (next(self.gen)), 3=user `with`.
                stacklevel=3,
            )

    @staticmethod
    def _validate_step_type(step_type: str) -> None:
        """Rejects a step_type that is empty, over-long, or holds control
        characters. It keys the rules, the web graph, and search, so it must be a
        short printable label — an empty one would also silently slip past the
        rule check. Arbitrary printable text (any language, symbols) is allowed."""
        if not isinstance(step_type, str) or not step_type.strip():
            raise ValueError("step_type must be a non-empty string.")
        if len(step_type) > _MAX_STEP_TYPE_LEN:
            raise ValueError(
                f"step_type is too long ({len(step_type)} characters; max "
                f"{_MAX_STEP_TYPE_LEN}). Use a short label for the kind of step."
            )
        if not step_type.isprintable():
            raise ValueError(
                "step_type must contain only printable characters "
                "(no newlines, tabs, or other control characters)."
            )

    def _resolve_parents(
        self, parent: Union["Node", str, List[Union["Node", str]], None]
    ) -> List["Node"]:
        """Resolves a `create_node` parent argument — None, a single Node/id, or a
        list of them (a join with several inputs) — to the list of parent nodes
        that exist in THIS store, de-duplicated and order-preserving.

        Raises ValueError if any supplied parent is not present in this store's
        index — a stale or foreign Node, or an unknown id — so a child can never
        be hung off a parent the store cannot resolve (which would otherwise only
        surface later as a KeyError from `get_lineage`).
        """
        if parent is None:
            return []
        if isinstance(parent, (Node, str)):
            parent = [parent]

        resolved: List["Node"] = []
        seen: Set[str] = set()
        for p in parent:
            if p is None:
                continue
            pid = p.node_id if isinstance(p, Node) else str(p)
            if pid in seen:
                continue
            if pid not in self.database.cache:
                raise ValueError(
                    f"Parent node {pid!r} is not present in this store. A parent "
                    "must be a node already created in or loaded from this store; "
                    "pass parent=None to create a root node."
                )
            seen.add(pid)
            resolved.append(self._node_from_index(pid))
        return resolved

    def _persist_if_touched(self, node: "Node", healthy: bool, duration: float) -> bool:
        """
        Persists and indexes the node if the user wrote any artifact or
        metadata, recording whether its code block ran to completion in the
        'healthy' flag, how long the block took in 'duration_s', and the
        total size of its files in 'size_mb'. Returns True if the node was
        persisted.

        When the store has dedupe enabled and the node completed cleanly, a
        node that is content-identical to an existing one is not persisted
        again: the existing node is reused (see `_deduplicate`) and True is
        returned without writing a second copy.
        """
        has_artifacts = bool(node.artifacts())
        has_user_meta = bool(set(node._hydrate()) - node._system_keys)
        if not (has_artifacts or has_user_meta):
            return False
        # Only deduplicate clean completions: a failed (unhealthy) run holds
        # partial work that should never be merged into a healthy node.
        if self.dedupe and healthy and self._deduplicate(node):
            return True
        size = sum(f.stat().st_size for f in node.path.rglob("*") if f.is_file())
        node._set_meta(
            "healthy", healthy, data_type="text", group="Structural Properties"
        )
        node._set_meta(
            "duration_s",
            round(duration, 3),
            data_type="text",
            group="Structural Properties",
        )
        node._set_meta(
            "size_mb",
            round(size / 1e6, 6),
            data_type="text",
            group="Structural Properties",
        )
        node._write_meta()
        self.database.add(node.node_id, node.to_db())
        # Sub-file deduplication happens off the write path: hand the node to the
        # background packer and return immediately, so the write costs ~a native
        # file write. Only clean completions are packed; a failed run's partial
        # files are left loose (and fully readable) on disk.
        if self.chunk and healthy:
            self._enqueue_pack(node.node_id)
        return True

    def _deduplicate(self, node: "Node") -> bool:
        """
        If `node` is content-identical to a node already in the store, rebinds
        it onto that existing node, discards the directory just written, and
        returns True. Otherwise stamps the node with its content hash so future
        runs can find it, and returns False.

        The content hash is only a fast bucket key: a candidate it points to is
        byte-verified with `_content_equal` before reuse, so a hash collision
        can never merge two genuinely different nodes.
        """
        content_hash = node.content_hash()
        candidate_id = self.database.find_by_hash(content_hash)
        if candidate_id:
            candidate = self.get_node(candidate_id)
            if candidate and node._content_equal(candidate):
                self._adopt(node, candidate)
                return True
        # New content (or an astronomically unlikely hash collision): record
        # the hash so a later identical node deduplicates against this one.
        node._set_meta(
            "content_hash",
            content_hash,
            data_type="text",
            group="Structural Properties",
        )
        return False

    @staticmethod
    def _adopt(node: "Node", existing: "Node") -> None:
        """
        Rebinds `node` in place onto an existing, content-identical node and
        deletes the directory `node` had just written. Because `create_node`
        yields this same object to the user, their `with ... as node` variable
        transparently becomes the existing node — including when later passed
        as a `parent`.
        """
        stale_dir = node.path
        node.node_id = existing.node_id
        node.path = existing.path
        node.generation = existing.generation
        node.parent_id = list(existing.parent_id)
        node.step_type = existing.step_type
        node._metadata = existing._hydrate()
        node._system_keys = set(existing._hydrate())
        if stale_dir.resolve() != existing.path.resolve():
            shutil.rmtree(stale_dir, ignore_errors=True)

    # ---------------------------------------------------------------------------
    # Deferred packing (off the write path)
    # ---------------------------------------------------------------------------

    def _enqueue_pack(self, node_id: str) -> None:
        """Hands a node to the background packer (starting it on first use)."""
        with self._pack_lock:
            if node_id in self._enqueued:
                return
            self._enqueued.add(node_id)
            if self._pack_worker is None or not self._pack_worker.is_alive():
                self._pack_worker = threading.Thread(
                    target=self._pack_loop, name="ancestree-packer", daemon=True
                )
                self._pack_worker.start()
            if not self._flush_registered:
                atexit.register(self.flush)  # converge on interpreter exit
                self._flush_registered = True
        self._pack_queue.put(node_id)

    def _pack_loop(self) -> None:
        """Background worker: chunks each queued node into the pool. Pure
        addition — it never deletes a loose file — so killing the process mid-run
        cannot lose data. Exceptions (e.g. a node pruned underneath it) are
        swallowed: the loose files stay readable and get packed next time."""
        self._scan_stragglers()  # pick up anything a previous run left loose
        while True:
            node_id = self._pack_queue.get()
            packed = False
            try:
                node = self.get_node(node_id)
                if node is not None:
                    node._pack()
                    packed = True
            except Exception:
                pass  # leave loose files in place; readable, packed next time
            finally:
                with self._pack_lock:
                    # This id is no longer queued/in-flight: drop it from the
                    # dedup set (which only needs pending work, so it stays
                    # bounded rather than growing for the life of the process).
                    self._enqueued.discard(node_id)
                    if packed:
                        self._reclaim_pending.add(node_id)
                self._pack_queue.task_done()

    def _scan_stragglers(self) -> None:
        """Enqueues any node still holding loose artifact files — e.g. left by a
        crash before the background packer reached it — so storage converges on
        the next run without a manual step. Runs once, in the worker thread."""
        try:
            self._scan_stragglers_inner()
        finally:
            self._scan_done.set()  # flush() waits on this before joining the queue

    def _scan_stragglers_inner(self) -> None:
        try:
            entries = list(self.root.iterdir())
        except OSError:
            return
        for entry in entries:
            if entry.name.startswith(".") or not (entry / "meta.json").exists():
                continue
            has_loose = any(
                f.is_file()
                and f.name not in ("meta.json", ARTIFACT_MANIFEST)
                and not f.name.endswith(".tmp")
                for f in entry.rglob("*")
            )
            if not has_loose:
                continue
            node = self.get_node(entry.name)
            if node is None:
                continue
            # A failed run's partial files are intentionally left loose and
            # unpacked; only converge healthy nodes a crash left behind.
            if node._hydrate().get("healthy", {}).get("value") is False:
                continue
            with self._pack_lock:
                if entry.name in self._enqueued:
                    continue
                self._enqueued.add(entry.name)
            self._pack_queue.put(entry.name)

    def flush(self) -> None:
        """Blocks until every queued artifact has been chunked, then reclaims the
        loose files whose recipe is now durable. This is the only place loose
        files are removed; it runs on the calling (main) thread, so no reader can
        be mid-read of a file it removes. Called automatically by the context
        manager and at interpreter exit; call it directly to converge sooner.

        Safe to interrupt: a partial flush just leaves more loose files to
        reclaim next time, and never loses data.
        """
        if not self.chunk:
            return
        with self._pack_lock:
            worker = self._pack_worker
        if worker is not None:
            # Make sure the worker has finished enqueuing any crash-stragglers
            # before we wait for the queue to drain, or we could return early.
            self._scan_done.wait()
        self._pack_queue.join()  # wait for the worker to chunk everything queued
        with self._pack_lock:
            pending = list(self._reclaim_pending)
            self._reclaim_pending.clear()
        for node_id in pending:
            node = self.get_node(node_id)
            if node is not None:
                node._reclaim_loose()

    # ---------------------------------------------------------------------------
    # Searching and Querying
    # ---------------------------------------------------------------------------

    def get_node(self, node: Union[str, "Node", None] = None) -> Optional["Node"]:
        """
        Resolves a node_id string into a Node object, loading it from disk.

        Accepts a Node as well (returned unchanged), so it can be used to normalise any "node or id" argument. Returns None rather than raising if the node does not exist or its metadata cannot be read.

        Args:
            node (Union[str, Node, None]): A node_id string, an existing Node instance, or None.

        Returns:
            Optional[Node]: The resolved Node, or None if the input is None, invalid, or not found.

        Examples:
            >>> node = store.get_node("abc12345")
        """
        if not node or str(node).lower() == "none":
            return None
        if isinstance(node, Node):
            return node
        node_path = self.root / node
        if not node_path.exists():
            return None
        try:
            return Node._load(node_path)
        except (FileNotFoundError, json.JSONDecodeError, AttributeError):
            return None

    def get_most_recent_node(self, **kwargs: Any) -> Optional["Node"]:
        """
        Finds the single most recently created node that matches the given search parameters.

        Recency is determined by the timestamp recorded when each node was created. Useful for picking up a pipeline where it left off, e.g. fetching the latest cleaned dataset.

        Args:
            **kwargs (Any): Key-value pairs to match against the nodes' searchable metadata keys, as in `find_node`.

        Returns:
            Optional[Node]: The most recent matching node, or None if nothing matches.

        Examples:
            >>> latest = store.get_most_recent_node(step_type="clean")
        """
        # find_matches is called here (not nested inside a database helper) so
        # a raising predicate warns at the same stacklevel as find_node.
        matches = self.database.find_matches(**kwargs)
        str_id = self.database.most_recent(matches)
        return self._node_from_index(str_id) if str_id else None

    def from_parent(self, node: Union[str, "Node"], filename: str) -> List[Path]:
        """
        Shortcut to get specific file(s) from the parent node of the specified node.

        Equivalent to looking up the node's parent and calling `artifacts` on it. The typical use is reading the previous step's output as the current step's input.

        Args:
            node (Union[str, Node]): The Node or node_id whose parent to search.
            filename (str): A glob pattern or substring to match against the parent's files, as in `Node.artifacts`.

        Returns:
            List[Path]: The matching file paths from the node's parent(s), ready to read directly (as returned by `Node.artifacts`). For a node with several parents the matches across all of them are returned, in parent order. Empty if the node or its parents cannot be found, the node has no parent, or nothing matches.

        Examples:
            >>> with store.create_node(step_type="model", parent=clean_node) as node:
            ...     [training_data] = store.from_parent(node, "cleaned.csv")
        """
        resolved = self.get_node(node)
        if resolved is None:
            return []
        matches: List[Path] = []
        for pid in resolved.parent_id:
            parent_node = self.get_node(pid)
            if parent_node:
                matches.extend(parent_node.artifacts(filename))
        return matches

    def find_node(self, **kwargs: Any) -> List["Node"]:
        """
        Search for nodes based on metadata key values. Values are matched by
        equality against the searchable metadata; pass a callable to express
        a predicate instead.

        Args:
            **kwargs (Any): Key-value pairs to match against the node's searchable metadata keys.

        Returns:
            List['Node']: A list of node objects that match all provided criteria.

        Examples:
            >>> store.find_node(step_type="ingest")
            >>> store.find_node(accuracy=lambda a: a is not None and a > 0.8)
        """
        return [
            self._node_from_index(node_id)
            for node_id in self.database.find_matches(**kwargs)
        ]

    def get_lineage(self, node: Union[str, "Node"]) -> List["Node"]:
        """
        Traces the ancestry of the node.

        Args:
            node (str | Node): The Node or node_id to trace from.

        Returns:
            List['Node']: A list of Node objects ordered from oldest ancestor to the target node.

        Examples:
            >>> history = store.get_lineage("abc12345")
            >>> [n.step_type for n in history]
            ['ingest', 'clean', 'transform']
        """
        if isinstance(node, Node):
            node = node.node_id
        return [
            self._node_from_index(node_id)
            for node_id in self.database.get_lineage(node)
        ]

    def find_in_lineage(self, node: Union[str, "Node"], **kwargs: Any) -> List["Node"]:
        """
        Searches a node's ancestry for nodes matching specified search parameters.

        This is `find_node` restricted to a single lineage: only the target node and its ancestors are considered. Values are matched by equality against the searchable metadata; pass a callable to express a predicate instead.

        Args:
            node (Union[str, Node]): The Node or node_id whose ancestry to search.
            **kwargs (Any): Key-value pairs to match against the nodes' searchable metadata keys.

        Returns:
            List[Node]: The nodes in the lineage that match all provided criteria.

        Examples:
            >>> store.find_in_lineage(model_node, step_type="clean")
        """
        if isinstance(node, Node):
            node = node.node_id
        return [
            self._node_from_index(node_id)
            for node_id in self.database.find_in_lineage(node, **kwargs)
        ]

    def get_child_nodes(self, node: Union[str, "Node"]) -> List["Node"]:
        """
        Returns the direct children of the specified node.

        Only immediate offspring are returned, not the full subtree. To walk further down the branch, call this on each child in turn.

        Args:
            node (Union[str, Node]): A Node object or node_id string.

        Returns:
            List[Node]: All nodes whose parent is the specified node. Empty if the node has no children or does not exist.
        """
        target = self.get_node(node)
        if not target:
            return []
        return [
            self._node_from_index(nid)
            for nid in self.database.find_children(target.node_id)
        ]

    def _node_from_index(self, node_id: str) -> "Node":
        """
        Builds a Node from the in-memory index without touching disk; the
        full metadata is hydrated lazily on first access. Only valid for ids
        present in the index.
        """
        return Node._from_index(self.root / node_id, self.database.cache[node_id])

    # ---------------------------------------------------------------------------
    # Maintenance
    # ---------------------------------------------------------------------------

    def rebuild_db_from_disk(self) -> None:
        """
        Rebuilds the search index by scanning all node directories on disk.

        Use this as a recovery step if the index becomes stale or corrupt —
        for example after a crash mid-write, manual filesystem changes, or a
        KeyError from get_lineage suggesting a missing index entry.

        Note: only nodes with a valid meta.json are re-indexed. Directories
        without one are silently skipped.
        """
        self.database.rebuild_from_disk()

    def prune(self, node: Union[str, "Node"], dry_run: bool = True) -> List["Node"]:
        """
        Deletes a node and the descendants it solely supports.

        A descendant is removed only when EVERY one of its parents is also being
        removed — a node still reachable from a branch you did not prune survives,
        with the pruned parent dropped from its `parent_id`. (For a plain tree
        this is just "delete the whole subtree".) Both the directories on disk and
        their index entries are removed. Run with the default `dry_run=True` first
        to preview exactly what would be removed.

        Args:
            node (Union[str, Node]): Either a node_id string or a Node object.
            dry_run (bool, optional): If True, returns the list of nodes that would be deleted without deleting anything. Must be set to False to actually delete. Defaults to True.

        Returns:
            List[Node]: Nodes that were (or would be) deleted, deepest first.

        Raises:
            PermissionError: If the target resolves to the store's root directory.

        Examples:
            >>> store.prune("abc12345")                  # preview only
            >>> store.prune("abc12345", dry_run=False)   # actually delete
        """
        deleted = self._prune(node, dry_run=dry_run)
        # Deleting nodes orphans the chunks only they referenced; reclaim them.
        if deleted and not dry_run and self.chunk:
            self.gc()
        return deleted

    def _prune(self, node: Union[str, "Node"], dry_run: bool) -> List["Node"]:
        target = self.get_node(node)
        if not target:
            return []

        if target.path.resolve() == self.root.resolve():
            raise PermissionError("Cannot prune the root lineageStore directory.")

        db = self.database
        target_id = target.node_id

        # 1. Collect the target and all its transitive descendants (the part of
        #    the DAG a deletion could touch), iteratively.
        affected: Set[str] = set()
        stack = [target_id]
        while stack:
            nid = stack.pop()
            if nid in affected:
                continue
            affected.add(nid)
            stack.extend(db.find_children(nid))

        # 2. Topologically order the affected nodes (a node after its affected
        #    parents) via Kahn's algorithm, then decide deletions in that order:
        #    the target goes, and a descendant goes only if ALL of its parents
        #    (anywhere) are being deleted. A surviving parent spares the child.
        indeg = {nid: 0 for nid in affected}
        for nid in affected:
            for p in db._parents(db.cache[nid]):
                if p in affected:
                    indeg[nid] += 1
        ready = [nid for nid in affected if indeg[nid] == 0]
        topo: List[str] = []
        while ready:
            nid = ready.pop()
            topo.append(nid)
            for child in db.find_children(nid):
                if child in affected:
                    indeg[child] -= 1
                    if indeg[child] == 0:
                        ready.append(child)

        deleted_ids: Set[str] = set()
        for nid in topo:
            if nid == target_id or all(
                p in deleted_ids for p in db._parents(db.cache[nid])
            ):
                deleted_ids.add(nid)

        # Deepest-first (children before parents) for the returned list.
        deleted = [
            self._node_from_index(nid) for nid in reversed(topo) if nid in deleted_ids
        ]
        if dry_run:
            return deleted

        for victim in deleted:
            shutil.rmtree(victim.path)
            db.remove(victim.node_id)

        # 3. Survivors keep their other parents but must drop the deleted ones,
        #    or they would point at nodes that no longer exist.
        for nid in affected - deleted_ids:
            survivor = self.get_node(nid)
            if survivor is None:
                continue
            kept = [p for p in survivor.parent_id if p not in deleted_ids]
            if kept != survivor.parent_id:
                survivor.parent_id = kept
                survivor._set_meta(
                    "parent_id", kept, data_type="text", group="Structural Properties"
                )
                survivor._write_meta()
                db.add(survivor.node_id, survivor.to_db())

        return deleted

    def gc(self) -> int:
        """
        Deletes chunks in the shared pool that no node references any more.

        Scans every node's artifact manifest for the chunks still in use, then
        removes the rest. A store-level lock makes concurrent collections safe,
        and chunks written in the last minute are spared so an in-flight pack in
        another process is never reaped before it records its recipe.

        Returns:
            int: The number of chunks deleted.
        """
        lock = self.root / ".gc.lock"
        try:
            fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            return 0  # another collection holds the lock; nothing to do
        try:
            live: set[str] = set()
            for entry in self.root.iterdir():
                manifest = entry / ARTIFACT_MANIFEST
                if not manifest.exists():
                    continue
                try:
                    records = json.loads(manifest.read_text())
                except json.JSONDecodeError:
                    continue
                for record in records.values():
                    live.update(record.get("chunks", ()))

            store = ChunkStore(self.root)
            grace = time.time() - 60
            removed = 0
            for digest in list(store.all_digests()):
                if digest not in live and store.mtime(digest) < grace:
                    store.delete(digest)
                    removed += 1
            return removed
        finally:
            os.close(fd)
            lock.unlink(missing_ok=True)

    # ---------------------------------------------------------------------------
    # Visualisation
    # ---------------------------------------------------------------------------

    def generate_web_graph(self) -> Path:
        """
        Creates an interactive web graph of node hierarchies and lineage.

        Renders every node in the store as a self-contained HTML file — all styles and scripts are inlined, so the file can be opened directly in a browser or shared as-is. Nodes are laid out by lineage and coloured by step type; clicking a node reveals its metadata and artifacts.

        The file is written to `<store root>/interactive_pipeline.html`, and the location is printed on completion.
        """
        path = run_web_generator(self)
        return path

__init__

__init__(root, rules=None, gen_triggers=None, dedupe=True, chunk=True)

Initialises the LineageStore, ensures its directory exists, and loads or creates the ruleset configuration.

On creation the LineageStore saves a .lineage_config.json file. On subsequent re-creation, the store reads from this file. There is no need to resupply rules or gen_triggers at any point after initial creation even if the store no longer exists in memory. The rules and gen_triggers cannot be changed after initial creation.

Parameters:

Name	Type	Description	Default
root	Union[Path, str]	Root directory for data pipeline. This is where the nodes sit.	required
rules	Dict	A mapping defining the allowed transitions. Defaults to None.	None
gen_triggers	List	Step types that mark a new generation. When a node of this type is created, its generation number increments by one relative to its parent. Defaults to None.	None
dedupe	bool	When True, a node that is content-identical to one already in the store is not created a second time: create_node reuses the existing node instead, and the variable yielded by the with block points at it. Two nodes are content-identical when they share the same step_type, the same parent, the same user metadata, and byte-identical artifacts; volatile fields (node_id, timestamp, duration, size) and provenance (user, platform, git state) are ignored. A candidate match is byte-verified before reuse. This is a behaviour of the store instance and is not persisted in the config — pass it each time you open a store you want to deduplicate. Defaults to False.	True
chunk	bool	When True, artifacts are deduplicated at sub-file granularity. As each node is persisted, its files are split into content-defined chunks stored once in a shared pool (<root>/.chunks); near-identical artifacts across nodes then cost only their differing chunks. This is transparent: you write files normally inside create_node, and reading them back (node / "file", node.artifacts(), the web graph) reassembles them on demand. Space is reclaimed by calling compact(). Like dedupe, this is a per-instance behaviour and is not persisted in the config. Defaults to False.	True

Examples:

>>> rules = {"clean": ["ingest"], "model": ["clean"]}
>>> store = LineageStore("my_project", rules=rules, gen_triggers=["ingest"])

Source code in src/ancestree/core.py

def __init__(
    self,
    root: Union[Path, str],
    rules: Optional[Dict[str, Any]] = None,
    gen_triggers: Optional[List[str]] = None,
    dedupe: bool = True,
    chunk: bool = True,
):
    """
    Initialises the LineageStore, ensures its directory exists, and loads or creates the ruleset configuration.

    On creation the LineageStore saves a .lineage_config.json file. On subsequent re-creation, the store reads from this file. There is no need to resupply rules or gen_triggers at any point after initial creation even if the store no longer exists in memory. The rules and gen_triggers cannot be changed after initial creation.

    Args:
        root (Union[Path, str]): Root directory for data pipeline. This is where the nodes sit.
        rules (Dict, optional): A mapping defining the allowed transitions. Defaults to None.
        gen_triggers (List, optional): Step types that mark a new generation. When a node of this type is created, its generation number increments by one relative to its parent. Defaults to None.
        dedupe (bool, optional): When True, a node that is content-identical to one already in the store is not created a second time: `create_node` reuses the existing node instead, and the variable yielded by the `with` block points at it. Two nodes are content-identical when they share the same step_type, the same parent, the same user metadata, and byte-identical artifacts; volatile fields (node_id, timestamp, duration, size) and provenance (user, platform, git state) are ignored. A candidate match is byte-verified before reuse. This is a behaviour of the store instance and is not persisted in the config — pass it each time you open a store you want to deduplicate. Defaults to False.
        chunk (bool, optional): When True, artifacts are deduplicated at sub-file granularity. As each node is persisted, its files are split into content-defined chunks stored once in a shared pool (`<root>/.chunks`); near-identical artifacts across nodes then cost only their differing chunks. This is transparent: you write files normally inside `create_node`, and reading them back (`node / "file"`, `node.artifacts()`, the web graph) reassembles them on demand. Space is reclaimed by calling `compact()`. Like `dedupe`, this is a per-instance behaviour and is not persisted in the config. Defaults to False.

    Examples:
        >>> rules = {"clean": ["ingest"], "model": ["clean"]}
        >>> store = LineageStore("my_project", rules=rules, gen_triggers=["ingest"])
    """
    self.root = Path(root)
    self.root.mkdir(parents=True, exist_ok=True)

    self.config_path = self.root / ".lineage_config.json"
    config = self._do_config(rules, gen_triggers)
    self.rules = config["rules"]
    self.triggers = config["triggers"]
    self.dedupe = dedupe
    self.chunk = chunk
    self.database = lineage_database(self.root)

    # Deferred packing: artifacts are written as normal files (native speed)
    # and chunked into the pool by a background worker, off the write path.
    # The worker only ever ADDS (chunks + manifest); loose files are removed
    # only by flush()/close, so an interrupt never loses data.
    self._pack_queue: "queue.Queue[str]" = queue.Queue()
    self._pack_worker: Optional[threading.Thread] = None
    self._pack_lock = threading.Lock()
    self._enqueued: Set[str] = set()  # node_ids queued for chunking
    self._reclaim_pending: Set[str] = set()  # chunked, awaiting loose reclaim
    self._flush_registered = False
    self._scan_done = threading.Event()  # worker finished its straggler scan
    # Track this store so the after-fork handler can re-arm its packer in a
    # forked child (a thread does not survive fork — see _reset_workers_after_fork).
    _live_stores.add(self)

clear_cache

clear_cache()

Discards the store's session read cache (<root>/.cache).

Reading a packed artifact reassembles its bytes into this cache rather than back into the node directory, so the store never needs compact() to reclaim what reads would otherwise leak. The cache is pure derived data — anything in it is regenerated from the chunk pool on the next read — so clearing it only frees disk, never loses anything. It is also cleared automatically when the process exits or a with block closes; call this to reclaim the space sooner.

Source code in src/ancestree/core.py

def clear_cache(self) -> None:
    """
    Discards the store's session read cache (`<root>/.cache`).

    Reading a packed artifact reassembles its bytes into this cache rather
    than back into the node directory, so the store never needs `compact()`
    to reclaim what reads would otherwise leak. The cache is pure derived
    data — anything in it is regenerated from the chunk pool on the next
    read — so clearing it only frees disk, never loses anything. It is also
    cleared automatically when the process exits or a `with` block closes;
    call this to reclaim the space sooner.
    """
    drop_read_cache(self.root)

create_node

create_node(step_type, parent=None)

Creates a new node while enforcing lineage rules.

The node only materialises on disk once the user writes an artifact or adds metadata; an untouched node is discarded with a warning. If the user's code raises after writing, the partial work is persisted and the node's 'healthy' metadata flag is set to False (True on clean completion).

If the store was opened with dedupe=True and the block completes cleanly, a node that is content-identical to one already in the store is not created again: the existing node is reused and the as node variable is rebound onto it. See LineageStore.__init__ for what counts as content-identical.

Parameters:

Name	Type	Description	Default
step_type	str	The type of pipeline step being performed.	required
parent	Node | str | list	The parent — a Node or node_id, or a list of them for a step that joins several inputs (a merge/join, making the lineage a DAG). Every parent must already exist in this store and individually satisfy the rules; the node's generation is one past its deepest parent. Defaults to None (a root). Examples: parent=clean, parent=[features_a, features_b].	None

Raises:

Type	Description
ValueError	If the step type transition is not permitted according to the store rules.

Yields:

Name	Type	Description
Node	Node	The new node, ready to receive artifacts and metadata.

Examples:

>>> with store.create_node(step_type="clean", parent=ingest_node) as node:
...     df.to_csv(node / "cleaned.csv")
...     node.add_meta("rows", len(df))

Source code in src/ancestree/core.py

@contextmanager
def create_node(
    self,
    step_type: str,
    parent: Union["Node", str, List[Union["Node", str]], None] = None,
) -> Iterator["Node"]:
    """Creates a new node while enforcing lineage rules.

    The node only materialises on disk once the user writes an artifact or
    adds metadata; an untouched node is discarded with a warning. If the
    user's code raises after writing, the partial work is persisted and the
    node's 'healthy' metadata flag is set to False (True on clean completion).

    If the store was opened with `dedupe=True` and the block completes
    cleanly, a node that is content-identical to one already in the store
    is not created again: the existing node is reused and the `as node`
    variable is rebound onto it. See `LineageStore.__init__` for what counts
    as content-identical.

    Args:
        step_type (str): The type of pipeline step being performed.
        parent (Node | str | list, optional): The parent — a Node or node_id, or a **list** of them for a step that joins several inputs (a merge/join, making the lineage a DAG). Every parent must already exist in this store and individually satisfy the rules; the node's generation is one past its deepest parent. Defaults to None (a root). Examples: `parent=clean`, `parent=[features_a, features_b]`.

    Raises:
        ValueError: If the step type transition is not permitted according to the store rules.

    Yields:
        Node: The new node, ready to receive artifacts and metadata.

    Examples:
        >>> with store.create_node(step_type="clean", parent=ingest_node) as node:
        ...     df.to_csv(node / "cleaned.csv")
        ...     node.add_meta("rows", len(df))
    """

    self._validate_step_type(step_type)
    parents = self._resolve_parents(parent)

    # Rule check: every parent must individually be a legal predecessor.
    allowed = self.rules.get(step_type)
    if allowed is not None:
        parent_types = [p.step_type for p in parents] or [None]
        for ptype in parent_types:
            if ptype not in allowed:
                raise ValueError(
                    f"Invalid transition: {ptype} -> {step_type}. "
                    f"Allowed parents: {allowed}."
                )

    # A node sits one generation below its deepest input (when it triggers a
    # new generation), otherwise at that depth.
    base_gen = max((p.generation for p in parents), default=0)
    current_gen = (
        base_gen + 1 if (parents and step_type in self.triggers) else base_gen
    )

    node_id = uuid.uuid4().hex[:8]
    while node_id in self.database.cache or (self.root / node_id).exists():
        node_id = uuid.uuid4().hex[:8]
    node_path = self.root / node_id

    parent_id = [p.node_id for p in parents]
    new_node = Node._create(
        node_path, node_id, current_gen, parent_id, step_type=step_type
    )

    start = time.monotonic()
    try:
        yield new_node
    except BaseException:
        # Keep partial work: anything written before the failure persists,
        # flagged as unhealthy. An untouched node leaves no trace.
        if not self._persist_if_touched(
            new_node, healthy=False, duration=time.monotonic() - start
        ):
            shutil.rmtree(new_node.path, ignore_errors=True)
        raise

    if not self._persist_if_touched(
        new_node, healthy=True, duration=time.monotonic() - start
    ):
        shutil.rmtree(new_node.path, ignore_errors=True)

        warnings.warn(
            f"Node '{new_node.node_id}' (step_type='{step_type}') was discarded: "
            "no artifacts were written and no metadata was added. "
            "Write at least one file or call node.add_meta() to persist the node.",
            UserWarning,
            # 1=here, 2=contextlib.__exit__ (next(self.gen)), 3=user `with`.
            stacklevel=3,
        )

find_in_lineage

find_in_lineage(node, **kwargs)

Searches a node's ancestry for nodes matching specified search parameters.

This is find_node restricted to a single lineage: only the target node and its ancestors are considered. Values are matched by equality against the searchable metadata; pass a callable to express a predicate instead.

Parameters:

Name	Type	Description	Default
node	Union[str, Node]	The Node or node_id whose ancestry to search.	required
**kwargs	Any	Key-value pairs to match against the nodes' searchable metadata keys.	{}

Returns:

Type	Description
List[Node]	List[Node]: The nodes in the lineage that match all provided criteria.

Examples:

>>> store.find_in_lineage(model_node, step_type="clean")

Source code in src/ancestree/core.py

def find_in_lineage(self, node: Union[str, "Node"], **kwargs: Any) -> List["Node"]:
    """
    Searches a node's ancestry for nodes matching specified search parameters.

    This is `find_node` restricted to a single lineage: only the target node and its ancestors are considered. Values are matched by equality against the searchable metadata; pass a callable to express a predicate instead.

    Args:
        node (Union[str, Node]): The Node or node_id whose ancestry to search.
        **kwargs (Any): Key-value pairs to match against the nodes' searchable metadata keys.

    Returns:
        List[Node]: The nodes in the lineage that match all provided criteria.

    Examples:
        >>> store.find_in_lineage(model_node, step_type="clean")
    """
    if isinstance(node, Node):
        node = node.node_id
    return [
        self._node_from_index(node_id)
        for node_id in self.database.find_in_lineage(node, **kwargs)
    ]

find_node

find_node(**kwargs)

Search for nodes based on metadata key values. Values are matched by equality against the searchable metadata; pass a callable to express a predicate instead.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Key-value pairs to match against the node's searchable metadata keys.	{}

Returns:

Type	Description
List[Node]	List['Node']: A list of node objects that match all provided criteria.

Examples:

>>> store.find_node(step_type="ingest")
>>> store.find_node(accuracy=lambda a: a is not None and a > 0.8)

Source code in src/ancestree/core.py

def find_node(self, **kwargs: Any) -> List["Node"]:
    """
    Search for nodes based on metadata key values. Values are matched by
    equality against the searchable metadata; pass a callable to express
    a predicate instead.

    Args:
        **kwargs (Any): Key-value pairs to match against the node's searchable metadata keys.

    Returns:
        List['Node']: A list of node objects that match all provided criteria.

    Examples:
        >>> store.find_node(step_type="ingest")
        >>> store.find_node(accuracy=lambda a: a is not None and a > 0.8)
    """
    return [
        self._node_from_index(node_id)
        for node_id in self.database.find_matches(**kwargs)
    ]

flush

flush()

Blocks until every queued artifact has been chunked, then reclaims the loose files whose recipe is now durable. This is the only place loose files are removed; it runs on the calling (main) thread, so no reader can be mid-read of a file it removes. Called automatically by the context manager and at interpreter exit; call it directly to converge sooner.

Safe to interrupt: a partial flush just leaves more loose files to reclaim next time, and never loses data.

Source code in src/ancestree/core.py

def flush(self) -> None:
    """Blocks until every queued artifact has been chunked, then reclaims the
    loose files whose recipe is now durable. This is the only place loose
    files are removed; it runs on the calling (main) thread, so no reader can
    be mid-read of a file it removes. Called automatically by the context
    manager and at interpreter exit; call it directly to converge sooner.

    Safe to interrupt: a partial flush just leaves more loose files to
    reclaim next time, and never loses data.
    """
    if not self.chunk:
        return
    with self._pack_lock:
        worker = self._pack_worker
    if worker is not None:
        # Make sure the worker has finished enqueuing any crash-stragglers
        # before we wait for the queue to drain, or we could return early.
        self._scan_done.wait()
    self._pack_queue.join()  # wait for the worker to chunk everything queued
    with self._pack_lock:
        pending = list(self._reclaim_pending)
        self._reclaim_pending.clear()
    for node_id in pending:
        node = self.get_node(node_id)
        if node is not None:
            node._reclaim_loose()

from_parent

from_parent(node, filename)

Shortcut to get specific file(s) from the parent node of the specified node.

Equivalent to looking up the node's parent and calling artifacts on it. The typical use is reading the previous step's output as the current step's input.

Parameters:

Name	Type	Description	Default
node	Union[str, Node]	The Node or node_id whose parent to search.	required
filename	str	A glob pattern or substring to match against the parent's files, as in Node.artifacts.	required

Returns:

Type	Description
List[Path]	List[Path]: The matching file paths from the node's parent(s), ready to read directly (as returned by Node.artifacts). For a node with several parents the matches across all of them are returned, in parent order. Empty if the node or its parents cannot be found, the node has no parent, or nothing matches.

Examples:

>>> with store.create_node(step_type="model", parent=clean_node) as node:
...     [training_data] = store.from_parent(node, "cleaned.csv")

Source code in src/ancestree/core.py

def from_parent(self, node: Union[str, "Node"], filename: str) -> List[Path]:
    """
    Shortcut to get specific file(s) from the parent node of the specified node.

    Equivalent to looking up the node's parent and calling `artifacts` on it. The typical use is reading the previous step's output as the current step's input.

    Args:
        node (Union[str, Node]): The Node or node_id whose parent to search.
        filename (str): A glob pattern or substring to match against the parent's files, as in `Node.artifacts`.

    Returns:
        List[Path]: The matching file paths from the node's parent(s), ready to read directly (as returned by `Node.artifacts`). For a node with several parents the matches across all of them are returned, in parent order. Empty if the node or its parents cannot be found, the node has no parent, or nothing matches.

    Examples:
        >>> with store.create_node(step_type="model", parent=clean_node) as node:
        ...     [training_data] = store.from_parent(node, "cleaned.csv")
    """
    resolved = self.get_node(node)
    if resolved is None:
        return []
    matches: List[Path] = []
    for pid in resolved.parent_id:
        parent_node = self.get_node(pid)
        if parent_node:
            matches.extend(parent_node.artifacts(filename))
    return matches

gc

gc()

Deletes chunks in the shared pool that no node references any more.

Scans every node's artifact manifest for the chunks still in use, then removes the rest. A store-level lock makes concurrent collections safe, and chunks written in the last minute are spared so an in-flight pack in another process is never reaped before it records its recipe.

Returns:

Name	Type	Description
int	int	The number of chunks deleted.

Source code in src/ancestree/core.py

def gc(self) -> int:
    """
    Deletes chunks in the shared pool that no node references any more.

    Scans every node's artifact manifest for the chunks still in use, then
    removes the rest. A store-level lock makes concurrent collections safe,
    and chunks written in the last minute are spared so an in-flight pack in
    another process is never reaped before it records its recipe.

    Returns:
        int: The number of chunks deleted.
    """
    lock = self.root / ".gc.lock"
    try:
        fd = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError:
        return 0  # another collection holds the lock; nothing to do
    try:
        live: set[str] = set()
        for entry in self.root.iterdir():
            manifest = entry / ARTIFACT_MANIFEST
            if not manifest.exists():
                continue
            try:
                records = json.loads(manifest.read_text())
            except json.JSONDecodeError:
                continue
            for record in records.values():
                live.update(record.get("chunks", ()))

        store = ChunkStore(self.root)
        grace = time.time() - 60
        removed = 0
        for digest in list(store.all_digests()):
            if digest not in live and store.mtime(digest) < grace:
                store.delete(digest)
                removed += 1
        return removed
    finally:
        os.close(fd)
        lock.unlink(missing_ok=True)

generate_web_graph

generate_web_graph()

Creates an interactive web graph of node hierarchies and lineage.

Renders every node in the store as a self-contained HTML file — all styles and scripts are inlined, so the file can be opened directly in a browser or shared as-is. Nodes are laid out by lineage and coloured by step type; clicking a node reveals its metadata and artifacts.

The file is written to <store root>/interactive_pipeline.html, and the location is printed on completion.

Source code in src/ancestree/core.py

def generate_web_graph(self) -> Path:
    """
    Creates an interactive web graph of node hierarchies and lineage.

    Renders every node in the store as a self-contained HTML file — all styles and scripts are inlined, so the file can be opened directly in a browser or shared as-is. Nodes are laid out by lineage and coloured by step type; clicking a node reveals its metadata and artifacts.

    The file is written to `<store root>/interactive_pipeline.html`, and the location is printed on completion.
    """
    path = run_web_generator(self)
    return path

get_child_nodes

get_child_nodes(node)

Returns the direct children of the specified node.

Only immediate offspring are returned, not the full subtree. To walk further down the branch, call this on each child in turn.

Parameters:

Name	Type	Description	Default
node	Union[str, Node]	A Node object or node_id string.	required

Returns:

Type	Description
List[Node]	List[Node]: All nodes whose parent is the specified node. Empty if the node has no children or does not exist.

Source code in src/ancestree/core.py

def get_child_nodes(self, node: Union[str, "Node"]) -> List["Node"]:
    """
    Returns the direct children of the specified node.

    Only immediate offspring are returned, not the full subtree. To walk further down the branch, call this on each child in turn.

    Args:
        node (Union[str, Node]): A Node object or node_id string.

    Returns:
        List[Node]: All nodes whose parent is the specified node. Empty if the node has no children or does not exist.
    """
    target = self.get_node(node)
    if not target:
        return []
    return [
        self._node_from_index(nid)
        for nid in self.database.find_children(target.node_id)
    ]

get_lineage

get_lineage(node)

Traces the ancestry of the node.

Parameters:

Name	Type	Description	Default
node	str | Node	The Node or node_id to trace from.	required

Returns:

Type	Description
List[Node]	List['Node']: A list of Node objects ordered from oldest ancestor to the target node.

Examples:

>>> history = store.get_lineage("abc12345")
>>> [n.step_type for n in history]
['ingest', 'clean', 'transform']

Source code in src/ancestree/core.py

def get_lineage(self, node: Union[str, "Node"]) -> List["Node"]:
    """
    Traces the ancestry of the node.

    Args:
        node (str | Node): The Node or node_id to trace from.

    Returns:
        List['Node']: A list of Node objects ordered from oldest ancestor to the target node.

    Examples:
        >>> history = store.get_lineage("abc12345")
        >>> [n.step_type for n in history]
        ['ingest', 'clean', 'transform']
    """
    if isinstance(node, Node):
        node = node.node_id
    return [
        self._node_from_index(node_id)
        for node_id in self.database.get_lineage(node)
    ]

get_most_recent_node

get_most_recent_node(**kwargs)

Finds the single most recently created node that matches the given search parameters.

Recency is determined by the timestamp recorded when each node was created. Useful for picking up a pipeline where it left off, e.g. fetching the latest cleaned dataset.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Key-value pairs to match against the nodes' searchable metadata keys, as in find_node.	{}

Returns:

Type	Description
Optional[Node]	Optional[Node]: The most recent matching node, or None if nothing matches.

Examples:

>>> latest = store.get_most_recent_node(step_type="clean")

Source code in src/ancestree/core.py

def get_most_recent_node(self, **kwargs: Any) -> Optional["Node"]:
    """
    Finds the single most recently created node that matches the given search parameters.

    Recency is determined by the timestamp recorded when each node was created. Useful for picking up a pipeline where it left off, e.g. fetching the latest cleaned dataset.

    Args:
        **kwargs (Any): Key-value pairs to match against the nodes' searchable metadata keys, as in `find_node`.

    Returns:
        Optional[Node]: The most recent matching node, or None if nothing matches.

    Examples:
        >>> latest = store.get_most_recent_node(step_type="clean")
    """
    # find_matches is called here (not nested inside a database helper) so
    # a raising predicate warns at the same stacklevel as find_node.
    matches = self.database.find_matches(**kwargs)
    str_id = self.database.most_recent(matches)
    return self._node_from_index(str_id) if str_id else None

get_node

get_node(node=None)

Resolves a node_id string into a Node object, loading it from disk.

Accepts a Node as well (returned unchanged), so it can be used to normalise any "node or id" argument. Returns None rather than raising if the node does not exist or its metadata cannot be read.

Parameters:

Name	Type	Description	Default
node	Union[str, Node, None]	A node_id string, an existing Node instance, or None.	None

Returns:

Type	Description
Optional[Node]	Optional[Node]: The resolved Node, or None if the input is None, invalid, or not found.

Examples:

>>> node = store.get_node("abc12345")

Source code in src/ancestree/core.py

def get_node(self, node: Union[str, "Node", None] = None) -> Optional["Node"]:
    """
    Resolves a node_id string into a Node object, loading it from disk.

    Accepts a Node as well (returned unchanged), so it can be used to normalise any "node or id" argument. Returns None rather than raising if the node does not exist or its metadata cannot be read.

    Args:
        node (Union[str, Node, None]): A node_id string, an existing Node instance, or None.

    Returns:
        Optional[Node]: The resolved Node, or None if the input is None, invalid, or not found.

    Examples:
        >>> node = store.get_node("abc12345")
    """
    if not node or str(node).lower() == "none":
        return None
    if isinstance(node, Node):
        return node
    node_path = self.root / node
    if not node_path.exists():
        return None
    try:
        return Node._load(node_path)
    except (FileNotFoundError, json.JSONDecodeError, AttributeError):
        return None

prune

prune(node, dry_run=True)

Deletes a node and the descendants it solely supports.

A descendant is removed only when EVERY one of its parents is also being removed — a node still reachable from a branch you did not prune survives, with the pruned parent dropped from its parent_id. (For a plain tree this is just "delete the whole subtree".) Both the directories on disk and their index entries are removed. Run with the default dry_run=True first to preview exactly what would be removed.

Parameters:

Name	Type	Description	Default
node	Union[str, Node]	Either a node_id string or a Node object.	required
dry_run	bool	If True, returns the list of nodes that would be deleted without deleting anything. Must be set to False to actually delete. Defaults to True.	True

Returns:

Type	Description
List[Node]	List[Node]: Nodes that were (or would be) deleted, deepest first.

Raises:

Type	Description
PermissionError	If the target resolves to the store's root directory.

Examples:

>>> store.prune("abc12345")                  # preview only
>>> store.prune("abc12345", dry_run=False)   # actually delete

Source code in src/ancestree/core.py

def prune(self, node: Union[str, "Node"], dry_run: bool = True) -> List["Node"]:
    """
    Deletes a node and the descendants it solely supports.

    A descendant is removed only when EVERY one of its parents is also being
    removed — a node still reachable from a branch you did not prune survives,
    with the pruned parent dropped from its `parent_id`. (For a plain tree
    this is just "delete the whole subtree".) Both the directories on disk and
    their index entries are removed. Run with the default `dry_run=True` first
    to preview exactly what would be removed.

    Args:
        node (Union[str, Node]): Either a node_id string or a Node object.
        dry_run (bool, optional): If True, returns the list of nodes that would be deleted without deleting anything. Must be set to False to actually delete. Defaults to True.

    Returns:
        List[Node]: Nodes that were (or would be) deleted, deepest first.

    Raises:
        PermissionError: If the target resolves to the store's root directory.

    Examples:
        >>> store.prune("abc12345")                  # preview only
        >>> store.prune("abc12345", dry_run=False)   # actually delete
    """
    deleted = self._prune(node, dry_run=dry_run)
    # Deleting nodes orphans the chunks only they referenced; reclaim them.
    if deleted and not dry_run and self.chunk:
        self.gc()
    return deleted

rebuild_db_from_disk

rebuild_db_from_disk()

Rebuilds the search index by scanning all node directories on disk.

Use this as a recovery step if the index becomes stale or corrupt — for example after a crash mid-write, manual filesystem changes, or a KeyError from get_lineage suggesting a missing index entry.

Note: only nodes with a valid meta.json are re-indexed. Directories without one are silently skipped.

Source code in src/ancestree/core.py

def rebuild_db_from_disk(self) -> None:
    """
    Rebuilds the search index by scanning all node directories on disk.

    Use this as a recovery step if the index becomes stale or corrupt —
    for example after a crash mid-write, manual filesystem changes, or a
    KeyError from get_lineage suggesting a missing index entry.

    Note: only nodes with a valid meta.json are re-indexed. Directories
    without one are silently skipped.
    """
    self.database.rebuild_from_disk()

---

Working with Nodes

You never construct a Node yourself — they are created by LineageStore.create_node and returned by the store's search and lineage methods. You will interact with them to read and attach metadata, locate artifacts, and build paths inside a node's directory.

ancestree.models.Node

Represents a single step in the pipeline: a directory on disk holding the step's artifacts and a meta.json describing it.

Nodes are not constructed directly. They are created by LineageStore.create_node and returned by the store's search and lineage methods (find_node, get_lineage, get_child_nodes, ...). Interact with a node to read and attach metadata, locate its artifacts, and build paths inside its directory with the / operator.

Attributes:

Name	Type	Description
path	Path	The node's directory on disk.
node_id	str	The unique 8-character identifier of the node.
generation	int	The generation number of the node in the pipeline.
parent_id	List[str]	The node_ids of this node's parent(s) — a list, since a node may join several inputs. Empty for a root node.
step_type	str	The type of pipeline step this node represents.

Source code in src/ancestree/models.py

class Node:
    """
    Represents a single step in the pipeline: a directory on disk holding the step's artifacts and a meta.json describing it.

    Nodes are not constructed directly. They are created by `LineageStore.create_node` and returned by the store's search and lineage methods (`find_node`, `get_lineage`, `get_child_nodes`, ...). Interact with a node to read and attach metadata, locate its artifacts, and build paths inside its directory with the `/` operator.

    Attributes:
        path (Path): The node's directory on disk.
        node_id (str): The unique 8-character identifier of the node.
        generation (int): The generation number of the node in the pipeline.
        parent_id (List[str]): The node_ids of this node's parent(s) — a list,
            since a node may join several inputs. Empty for a root node.
        step_type (str): The type of pipeline step this node represents.
    """

    def __init__(
        self,
        path: Path,
        node_id: str,
        generation: int,
        parent_id: Optional[List[str]] = None,
        step_type: Optional[str] = None,
    ):
        """
        Initialises a node instance. Performs no I/O: use Node._load() to read
        an existing node from disk or Node._create() to initialise a new one.

        Args:
            path (Path): The filesystem path where the node is stored.
            node_id (str): A unique 8-character alphanumeric identifier.
            generation (int): The generation number of the node in the pipeline.
            parent_id (List[str]): The ids of the node's parent(s); empty/None for a root.
        """
        self.path = path
        self.node_id = node_id
        self.generation = generation
        self.parent_id: List[str] = list(parent_id) if parent_id else []
        self.step_type = step_type
        self._metadata: Optional[Dict[str, Any]] = {}
        self._system_keys: set[Any] = set()
        # Lazily loaded recipe (relpath -> {size, sha256, chunks}) for artifacts
        # that have been packed into the chunk store. Empty when nothing is packed.
        self._manifest_data: Optional[Dict[str, Any]] = None

    @property
    def metadata(self) -> Dict[str, Dict[str, Any]]:
        """
        The node's full metadata as a dictionary.

        Each key maps to an entry of the form `{'value': ..., 'type': ..., 'group': ..., 'searchable': ...}`. This includes both the structural metadata written by the store (node_id, parent_id, generation, step_type, timestamp, provenance) and anything added with `add_meta`.

        Returns a deep copy: mutating the result does not change the node. Use `add_meta` to modify metadata.

        Returns:
            Dict[str, Dict]: A mapping of metadata key to its entry dictionary.

        Examples:
            >>> node.metadata["accuracy"]["value"]
            0.92
        """
        return deepcopy(self._hydrate())

    def _hydrate(self) -> Dict[str, Any]:
        # Index-backed nodes (_from_index) defer reading meta.json until the
        # full metadata is actually needed.
        if self._metadata is None:
            self._metadata = json.loads((self.path / "meta.json").read_text())
        assert self._metadata is not None
        return self._metadata

    @classmethod
    def _from_index(cls, path: Path, flat: Dict[str, Any]) -> "Node":
        """
        Builds a node from the database's flattened index entry without
        touching disk. The full metadata is hydrated lazily from meta.json
        on first access.

        Args:
            path (Path): The directory of the node.
            flat (dict): The node's flat index entry (key -> value).

        Returns:
            Node: The index-backed node.
        """
        node = cls(
            path,
            flat.get("node_id", ""),
            flat.get("generation", 0),
            flat.get("parent_id"),
            step_type=flat.get("step_type"),
        )
        node._metadata = None
        return node

    @classmethod
    def _load(cls, path: Path) -> "Node":
        """
        Loads an existing node from its meta.json, which is the single source
        of truth for the structural attributes.

        Args:
            path (Path): The directory of the node to load.

        Returns:
            Node: The loaded node.
        """
        meta = json.loads((path / "meta.json").read_text())

        def _value(key: str) -> Any:
            entry = meta.get(key)
            return entry.get("value") if entry else None

        node = cls(
            path,
            _value("node_id"),
            _value("generation"),
            _value("parent_id"),
            step_type=_value("step_type"),
        )
        node._metadata = meta
        return node

    @classmethod
    def _create(
        cls,
        path: Path,
        node_id: str,
        generation: int,
        parent_id: Optional[List[str]] = None,
        step_type: Optional[str] = None,
    ) -> "Node":
        """
        Initialises a brand new node with its structural and provenance
        metadata. The initial keys are recorded in _system_keys so the store
        can tell system-written metadata apart from user additions.

        Args:
            path (Path): The filesystem path where the node is stored.
            node_id (str): A unique 8-character alphanumeric identifier.
            generation (int): The generation number of the node in the pipeline.
            parent_id (List[str]): The ids of the node's parent(s).

        Returns:
            Node: The new node. Nothing is written to disk until _write_meta().
        """
        node = cls(path, node_id, generation, parent_id, step_type=step_type)
        node._set_meta(
            "node_id", node_id, data_type="text", group="Structural Properties"
        )
        # Stored as a searchable list (data_type 'text' keeps it in the index, so
        # the database can find a node's children across any of its parents).
        node._set_meta(
            "parent_id", node.parent_id, data_type="text", group="Structural Properties"
        )
        node._set_meta(
            "generation", generation, data_type="text", group="Structural Properties"
        )
        node._set_meta(
            "step_type", step_type, data_type="text", group="Structural Properties"
        )
        node._set_meta(
            "timestamp",
            datetime.now(timezone.utc).isoformat(),
            data_type="text",
            group="Structural Properties",
        )
        for key, value in get_provenance().items():
            node._set_meta(
                key, value, data_type="text", group="Provenance", searchable=False
            )
        node._system_keys = set(node._hydrate())
        return node

    #: Path suffixes rendered inline as images when the data_type is inferred.
    IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".gif", ".svg", ".webp"}

    @staticmethod
    def _infer_data_type(value: Any) -> _DataType:
        """
        Maps a value to its natural data_type for 'auto' mode. Only types
        that are unambiguous signals are inferred: DataFrames, dicts/lists,
        Path objects (a Path is a deliberate file reference, never prose),
        and URL strings. Plain strings always stay 'text' — sniffing string
        contents is how auto-detection produces surprises.
        """
        if is_pandas(value):
            return "table"
        if isinstance(value, (dict, list)):
            return "json"
        if isinstance(value, Path):
            return "image" if value.suffix.lower() in Node.IMAGE_SUFFIXES else "link"
        if isinstance(value, str) and value.startswith(("http://", "https://")):
            return "link"
        return "text"

    def add_meta(
        self,
        key: str,
        value: Any,
        group: Optional[str] = "General",
        data_type: _DataType = "auto",
        searchable: bool = True,
    ) -> None:
        """
        Attaches a piece of metadata to the node.

        Metadata is what makes a node discoverable: every searchable entry can be matched by the store's search methods (`find_node`, `find_in_lineage`, `get_most_recent_node`) and is displayed in the interactive web graph. Adding a key that already exists overwrites the previous entry.

        Args:
            key (str): The name of the metadata entry.
            value (Any): The value to store. Must be JSON-serialisable.
            group (str, optional): A heading to group related entries under in the web graph display. Defaults to General.
            data_type (str, optional): How the value is rendered in the web graph. The default 'auto' infers it from the value: pandas DataFrames render as tables, dicts and lists as formatted JSON, Path objects as inline images (image suffixes) or file links, http(s) strings as links, and everything else as plain text. Pass a type explicitly to override the inference: 'image' for a path to an image file inside the node (the value is rewritten relative to the store root), 'link' for a clickable link, 'table' for a pandas DataFrame (stored as columns/rows), 'json' for a dict or list, 'code' for a monospaced snippet (e.g. the SQL or shell command that produced the node), or 'text' for plain text.
            searchable (bool, optional): If True the entry is indexed and can be matched by the store's search methods. Set to False for display-only metadata. Defaults to True.

        Examples:
            >>> with store.create_node(step_type="model", parent=clean_node) as node:
            ...     # Plain values — searchable by default
            ...     node.add_meta("accuracy", 0.94, group="Metrics")
            ...     node.add_meta("epochs", 42, group="Metrics")
            ...     node.add_meta("learning_rate", 1e-3, group="Config")
            ...
            ...     # Dict/list — rendered as formatted JSON in the web graph
            ...     node.add_meta("params", {"optimizer": "adam", "dropout": 0.3}, group="Config")
            ...
            ...     # Table — pandas DataFrame rendered as a sortable table; not searchable
            ...     node.add_meta("results", df, data_type="table", group="Outputs")
            ...
            ...     # Image — path rewritten relative to store root, rendered inline
            ...     fig.savefig(node / "confusion.png")
            ...     node.add_meta("confusion_matrix", node / "confusion.png", group="Figures")
            ...
            ...     # Code — rendered in a monospaced block
            ...     node.add_meta("query", "SELECT * FROM runs WHERE status = 'ok'", data_type="code")
            ...
            ...     # Display-only — visible in the web graph but excluded from find_node
            ...     node.add_meta("notes", "rerun after fixing label encoding", searchable=False)
            ...
            ...     # External link — rendered as a clickable URL
            ...     node.add_meta("wandb_run", "https://wandb.ai/my-org/run/abc123", group="Links")
        """
        if key in _RESERVED_KEYS:
            raise ValueError(
                f"{key!r} is a reserved key set by the store and cannot be "
                "set via add_meta."
            )
        self._set_meta(
            key, value, group=group, data_type=data_type, searchable=searchable
        )

    def _set_meta(
        self,
        key: str,
        value: Any,
        group: Optional[str] = "General",
        data_type: _DataType = "auto",
        searchable: bool = True,
    ) -> None:
        """Write a metadata entry, bypassing the reserved-key guard.

        The store calls this to set its own structural and provenance keys —
        the ones `add_meta` refuses from users. Validation, type inference and
        coercion are shared; only the guard is not.
        """
        if data_type not in _VALID_DATA_TYPES:
            raise ValueError(
                f"Invalid data_type {data_type!r}. "
                f"Must be one of: {', '.join(sorted(_VALID_DATA_TYPES))}"
            )

        if data_type == "auto":
            data_type = self._infer_data_type(value)

        if data_type in ("image", "link") and not str(value).startswith(
            ("http://", "https://")
        ):
            # Store file references relative to the store root so they
            # resolve from the generated HTML. URLs pass through untouched —
            # Path() would collapse their double slash.
            value = str(
                Path(
                    str(value)
                    .removeprefix(str(self.path.parent) + "/")
                    .removeprefix(str(self.path.parent))
                )
            )
            searchable = False

        if data_type == "table":
            if not is_pandas(value):
                raise TypeError(
                    f"Expected a pandas DataFrame for 'table', got {type(value).__name__}"
                )
            split = value.to_dict(orient="split")
            value = {"columns": split["columns"], "rows": split["data"]}
            searchable = False

        if data_type == "json":
            if not isinstance(value, (dict, list)):
                raise TypeError(
                    f"Expected a dict or list for 'json', got {type(value).__name__}"
                )
            searchable = False

        # Coerce numpy/pandas and other common non-JSON types to native Python so
        # the eventual meta.json write cannot fail, and warn that we did. Anything
        # still not serialisable is rejected here, at the call site, rather than
        # at block exit where the traceback would not point at this add_meta.
        value, coerced = to_jsonable(value)
        if coerced:
            warnings.warn(
                f"Metadata {key!r} held values that are not natively JSON-"
                "serialisable (e.g. numpy/pandas scalars, arrays, sets, "
                "datetimes); they were coerced to plain Python types. Pass native "
                "Python types to silence this.",
                UserWarning,
                stacklevel=3,
            )
        try:
            json.dumps(value)
        except (TypeError, ValueError) as exc:
            raise TypeError(
                f"Metadata {key!r} is not JSON-serialisable even after coercion "
                f"({type(exc).__name__}: {exc}). Convert it to plain Python types "
                "(int/float/str/bool/list/dict) before calling add_meta."
            ) from None

        entry = {
            f"{key}": {
                "value": value,
                "data_type": data_type,
                "group": group,
                "searchable": searchable,
            }
        }

        self._hydrate().update(entry)

    def _write_meta(self) -> None:
        """
        Internal helper for creating and writing metadata atomically to prevent corruption during crashes.
        """
        self.path.mkdir(parents=True, exist_ok=True)
        # Atomic write
        try:
            temp_file = self.path / "meta.json.tmp"
            temp_file.write_text(json.dumps(self.metadata, indent=2))
            temp_file.replace(self.path / "meta.json")
        finally:
            if temp_file.exists():
                temp_file.unlink()

    def _content_meta(self) -> Dict[str, Any]:
        """The node's user-supplied metadata entries — everything that is not
        a structural, provenance, or content-hash key. This is the metadata
        portion of the node's content identity."""
        return {
            key: self._normalise_entry(entry)
            for key, entry in self._hydrate().items()
            if key not in _NON_CONTENT_KEYS
        }

    def _normalise_entry(self, entry: Dict[str, Any]) -> Dict[str, Any]:
        """Strips this node's id out of file-referencing metadata so identical
        content fingerprints the same. An ``image``/``link`` value that points
        at a file inside the node is stored with the (random) node_id as a path
        segment; left in, it would make every run look unique. URLs are real
        content and pass through untouched. Returns a copy when rewritten; the
        node's own metadata is never mutated."""
        if entry.get("data_type") not in ("image", "link"):
            return entry
        value = entry.get("value")
        if not isinstance(value, str) or value.startswith(("http://", "https://")):
            return entry
        parts = Path(value).parts
        if self.node_id not in parts:
            return entry
        rest = parts[parts.index(self.node_id) + 1 :]
        return {**entry, "value": Path(*rest).as_posix() if rest else ""}

    def _artifact_digests(self) -> Dict[str, str]:
        """Maps each artifact's path (relative to the node's *own* directory,
        so it is independent of the random node_id) to the SHA-256 of its
        bytes. For packed artifacts the digest is read straight from the recipe,
        so no reassembly is needed. meta.json and the manifest are excluded —
        they describe the node, they are not its content."""
        manifest = self._manifest()
        digests: Dict[str, str] = {
            rel: record["sha256"] for rel, record in manifest.items()
        }
        for f in sorted(self.path.rglob("*")):
            if f.is_file() and f.name not in ("meta.json", ARTIFACT_MANIFEST):
                rel = f.relative_to(self.path).as_posix()
                digests.setdefault(rel, hashlib.sha256(f.read_bytes()).hexdigest())
        return digests

    def content_hash(self) -> str:
        """A SHA-256 fingerprint of the node's content: its step_type, parent,
        user metadata, and the bytes of every artifact. Excludes volatile
        fields (node_id, timestamp, duration, size, healthy flag) and
        provenance, so two runs producing the same step with the same metadata
        and identical artifact bytes share a fingerprint.

        Used by the store to deduplicate nodes when `dedupe=True`. The hash is
        only a fast bucket key — the store byte-verifies a candidate match with
        `_content_equal` before reusing it.

        Returns:
            str: The hex SHA-256 digest.
        """
        payload = {
            "step_type": self.step_type,
            "parents": sorted(self.parent_id),  # order-independent
            "meta": self._content_meta(),
            "artifacts": self._artifact_digests(),
        }
        blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
        return hashlib.sha256(blob.encode("utf-8")).hexdigest()

    def _content_equal(self, other: "Node") -> bool:
        """True if this node and `other` are content-identical: same step_type,
        same parent, same user metadata, and byte-identical artifacts. This is
        the definitive check behind deduplication — the content hash only
        narrows the candidates; equality is confirmed here so a hash collision
        can never cause two genuinely different nodes to be merged."""
        return (
            self.step_type == other.step_type
            and set(self.parent_id) == set(other.parent_id)
            and self._content_meta() == other._content_meta()
            and self._artifact_digests() == other._artifact_digests()
        )

    def to_db(self) -> Dict[str, Any]:
        # Flat {searchable key: value} dict for indexing/searching. Shares the
        # one definition of "searchable flattening" with the database, which
        # applies it to raw meta.json dicts on reconcile/rebuild.
        return flatten_meta(self._hydrate())

    def artifacts(self, contains: str = "*") -> List[Path]:
        """
        Searches this node's directory returning all files excluding internal metadata.
        Recursively finds all artifacts regardless of storage depth.

        Args:
            contains (str, optional): A glob pattern to filter discovered files. A plain substring also works: it is matched anywhere in the filename, case-insensitively. Defaults to "*" (all files).

        Returns:
            List[Path]: The matching file paths, ready to read or pass to a
                loader directly (no need to prefix the store root). A loose file
                points inside the node's directory; a packed artifact points at a
                reassembled copy in the store's read cache. Either way the path
                holds readable bytes, mirroring what `node / "file"` returns.

        Examples:
            >>> node.artifacts("*.csv")
            [PosixPath('/data/store/abc12345/sample.csv')]
            >>> pd.read_csv(node.artifacts("sample")[0])
        """
        search_pattern = contains
        if "*" not in contains and "?" not in contains:
            search_pattern = f"*{contains}*"

        artifacts = []
        for rel in sorted(self._artifact_rels()):
            name = rel.rsplit("/", 1)[-1]
            if (
                Path(rel).match(search_pattern)
                or name.lower().find(contains.lower()) != -1
            ):
                artifacts.append(self._resolve(rel))
        return artifacts

    def _artifact_rels(self) -> Set[str]:
        """The node's logical artifact paths (relative to the node): the union of
        packed manifest entries and loose files actually on disk. meta.json and
        the manifest are excluded — they describe the node, they are not its
        content. This is the storage-independent view artifact listing and the
        web graph build on, since a packed artifact has no file in the node dir."""
        rels = set(self._manifest())
        for f in self.path.rglob("*"):
            if f.is_file() and f.name not in ("meta.json", ARTIFACT_MANIFEST):
                rels.add(f.relative_to(self.path).as_posix())
        return rels

    def _resolve(self, rel: str) -> Path:
        """Returns a readable filesystem path for the logical artifact `rel`.

        Prefers the loose file when it is still on disk — that is a native read,
        no decompression — which is the common case for artifacts written this
        session before the background packer has reclaimed them. Once the loose
        file is gone the artifact is served from its chunks via the read cache;
        the manifest is reloaded first in case it was packed after we cached it.
        Because a loose file is only ever removed once its recipe is durable, a
        missing loose file guarantees the manifest can rebuild it."""
        loose = self.path / rel
        if loose.exists():
            return loose
        if rel not in self._manifest():
            self._manifest_data = None  # may have been packed since we cached it
        if rel in self._manifest():
            return self._materialize(rel)
        return loose

    def __truediv__(self, relative_loc: Union[Path, str]) -> Path:
        """
        Allows the use of the '/' operator to create paths inside the node's directory, mirroring pathlib.
        This is the idiomatic way to choose where to write an artifact. Any intermediate directories in the path are created automatically, so the returned path is always ready to write to.

        Args:
            relative_loc (Union[Path, str]): The string or Path object to append to the node's base path.

        Returns:
            Path: The resolved destination inside the node's directory.

        Raises:
            ValueError: If the path escapes the node's directory.

        Examples:
            >>> with store.create_node(step_type="clean") as node:
            ...     df.to_csv(node / "results/cleaned.csv")
        """
        target_path = (self.path / relative_loc).resolve()
        if not target_path.is_relative_to(self.path.resolve()):
            raise ValueError(
                f"Artifact path {relative_loc!r} escapes the node directory. "
                "Keep all artifact paths inside the node."
            )
        # A packed artifact with no loose file on disk is a read: reassemble it
        # into the read cache and return that path (reading is transparent). A
        # loose file (read) or a not-yet-written path (write target) returns a
        # path inside the node directory.
        rel = target_path.relative_to(self.path.resolve()).as_posix()
        if not target_path.exists():
            if rel not in self._manifest():
                self._manifest_data = None  # may have been packed in the background
            if rel in self._manifest():
                return self._materialize(rel)
        target_path.parent.mkdir(parents=True, exist_ok=True)
        return target_path

    # ---------------------------------------------------------------------------
    # Chunked storage (manifest, packing, reassembly)
    # ---------------------------------------------------------------------------

    def _manifest(self) -> Dict[str, Any]:
        """The recipe for every packed artifact, loaded lazily and cached.
        Empty when nothing in this node is packed."""
        if self._manifest_data is None:
            path = self.path / ARTIFACT_MANIFEST
            self._manifest_data = json.loads(path.read_text()) if path.exists() else {}
        return self._manifest_data

    def _write_manifest(self, manifest: Dict[str, Any]) -> None:
        self._manifest_data = manifest
        path = self.path / ARTIFACT_MANIFEST
        tmp = self.path / f"{ARTIFACT_MANIFEST}.tmp"
        try:
            tmp.write_text(json.dumps(manifest, indent=2))
            tmp.replace(path)
        finally:
            if tmp.exists():
                tmp.unlink()

    def _pack(self) -> None:
        """Chunks every loose artifact file into the shared pool and records the
        recipe in the manifest. This only ever ADDS — it writes content-addressed
        chunks (idempotent: an already-present chunk is a no-op) and then
        atomically replaces the manifest. It never deletes a loose file; that is
        done separately by `_reclaim_loose` at a quiescent point.

        Because of that ordering it is safe to interrupt at any instant (Ctrl+C,
        crash, kill): an artifact is always recoverable from its loose file (here
        until reclaimed) AND, once this returns, from its chunks — there is never
        a moment where neither exists. A half-finished run leaves only orphan
        chunks (reclaimed later by `gc`) and an unchanged manifest."""
        store = ChunkStore(self.path.parent)
        manifest = dict(self._manifest())
        changed = False
        for f in sorted(self.path.rglob("*")):
            if not f.is_file() or f.name in ("meta.json", ARTIFACT_MANIFEST):
                continue
            if f.name.endswith(".tmp"):
                continue
            rel = f.relative_to(self.path).as_posix()
            if rel in manifest:
                continue  # already chunked (artifacts are immutable)
            data = f.read_bytes()
            manifest[rel] = {
                "size": len(data),
                "sha256": hashlib.sha256(data).hexdigest(),
                "chunks": [store.put(chunk) for chunk in chunk_bytes(data)],
            }
            changed = True
        if changed:
            self._write_manifest(manifest)  # durable before any loose file is removed

    def _reclaim_loose(self) -> None:
        """Removes loose artifact files whose recipe is already durably in the
        manifest, reclaiming the space the chunk pool now holds. Call ONLY at a
        quiescent point (no reader may be holding a loose path) — `flush`/close
        do. Safe to interrupt: every file removed is reproducible from its
        chunks, so a partial run just leaves more to reclaim next time."""
        manifest = self._manifest()
        if not manifest:
            return
        for f in sorted(self.path.rglob("*")):
            if not f.is_file() or f.name in ("meta.json", ARTIFACT_MANIFEST):
                continue
            if f.name.endswith(".tmp"):
                continue
            rel = f.relative_to(self.path).as_posix()
            if rel in manifest:
                f.unlink(missing_ok=True)

    def _materialize(self, rel: str) -> Path:
        """Reassembles a packed artifact from its chunks into the store's read
        cache and returns that path. A cached copy from earlier in the session is
        reused as-is; otherwise the chunks are read and decompressed (in parallel
        for large files), verified against the recorded SHA-256, and written
        atomically. The node directory is never touched, so packed nodes stay
        packed without a manual `compact`."""
        record = self._manifest()[rel]
        out = get_read_cache(self.path.parent).path_for(self.node_id, rel)
        if out.exists():
            return out  # already reassembled this session

        store = ChunkStore(self.path.parent)
        digests = record["chunks"]
        if len(digests) >= _MATERIALIZE_THREAD_THRESHOLD and _MATERIALIZE_WORKERS > 1:
            # store.get reads + zlib-decompresses each chunk, both releasing the
            # GIL, so the pool overlaps real work. map() keeps chunk order.
            with ThreadPoolExecutor(max_workers=_MATERIALIZE_WORKERS) as pool:
                data = b"".join(pool.map(store.get, digests, chunksize=16))
        else:
            data = b"".join(store.get(h) for h in digests)

        if hashlib.sha256(data).hexdigest() != record["sha256"]:
            raise RuntimeError(
                f"Artifact {rel!r} failed its integrity check while being "
                "reassembled — the chunk store may be corrupt."
            )
        # Atomic publish: a unique temp keeps concurrent reassembles of the same
        # entry from clobbering each other mid-write.
        tmp = out.with_name(f"{out.name}.{uuid.uuid4().hex}.tmp")
        tmp.write_bytes(data)
        tmp.replace(out)
        return out

    def __repr__(self) -> str:
        """
        Returns a developer friendly string representation of the node.

        Examples:
            >>> node = store.get_node("abc12345")
            >>> print(node)
            'Node = abc12345, path = abc12345, generation = 0'
        """
        return f"Node = {self.node_id}, path = {self.path.name}, generation = {self.generation}"

metadata property

The node's full metadata as a dictionary.

Each key maps to an entry of the form {'value': ..., 'type': ..., 'group': ..., 'searchable': ...}. This includes both the structural metadata written by the store (node_id, parent_id, generation, step_type, timestamp, provenance) and anything added with add_meta.

Returns a deep copy: mutating the result does not change the node. Use add_meta to modify metadata.

Returns:

Type	Description
Dict[str, Dict[str, Any]]	Dict[str, Dict]: A mapping of metadata key to its entry dictionary.

Examples:

>>> node.metadata["accuracy"]["value"]
0.92

add_meta(key, value, group='General', data_type='auto', searchable=True)

Attaches a piece of metadata to the node.

Metadata is what makes a node discoverable: every searchable entry can be matched by the store's search methods (find_node, find_in_lineage, get_most_recent_node) and is displayed in the interactive web graph. Adding a key that already exists overwrites the previous entry.

Parameters:

Name	Type	Description	Default
key	str	The name of the metadata entry.	required
value	Any	The value to store. Must be JSON-serialisable.	required
group	str	A heading to group related entries under in the web graph display. Defaults to General.	'General'
data_type	str	How the value is rendered in the web graph. The default 'auto' infers it from the value: pandas DataFrames render as tables, dicts and lists as formatted JSON, Path objects as inline images (image suffixes) or file links, http(s) strings as links, and everything else as plain text. Pass a type explicitly to override the inference: 'image' for a path to an image file inside the node (the value is rewritten relative to the store root), 'link' for a clickable link, 'table' for a pandas DataFrame (stored as columns/rows), 'json' for a dict or list, 'code' for a monospaced snippet (e.g. the SQL or shell command that produced the node), or 'text' for plain text.	'auto'
searchable	bool	If True the entry is indexed and can be matched by the store's search methods. Set to False for display-only metadata. Defaults to True.	True

Examples:

>>> with store.create_node(step_type="model", parent=clean_node) as node:
...     # Plain values — searchable by default
...     node.add_meta("accuracy", 0.94, group="Metrics")
...     node.add_meta("epochs", 42, group="Metrics")
...     node.add_meta("learning_rate", 1e-3, group="Config")
...
...     # Dict/list — rendered as formatted JSON in the web graph
...     node.add_meta("params", {"optimizer": "adam", "dropout": 0.3}, group="Config")
...
...     # Table — pandas DataFrame rendered as a sortable table; not searchable
...     node.add_meta("results", df, data_type="table", group="Outputs")
...
...     # Image — path rewritten relative to store root, rendered inline
...     fig.savefig(node / "confusion.png")
...     node.add_meta("confusion_matrix", node / "confusion.png", group="Figures")
...
...     # Code — rendered in a monospaced block
...     node.add_meta("query", "SELECT * FROM runs WHERE status = 'ok'", data_type="code")
...
...     # Display-only — visible in the web graph but excluded from find_node
...     node.add_meta("notes", "rerun after fixing label encoding", searchable=False)
...
...     # External link — rendered as a clickable URL
...     node.add_meta("wandb_run", "https://wandb.ai/my-org/run/abc123", group="Links")

Source code in src/ancestree/models.py

def add_meta(
    self,
    key: str,
    value: Any,
    group: Optional[str] = "General",
    data_type: _DataType = "auto",
    searchable: bool = True,
) -> None:
    """
    Attaches a piece of metadata to the node.

    Metadata is what makes a node discoverable: every searchable entry can be matched by the store's search methods (`find_node`, `find_in_lineage`, `get_most_recent_node`) and is displayed in the interactive web graph. Adding a key that already exists overwrites the previous entry.

    Args:
        key (str): The name of the metadata entry.
        value (Any): The value to store. Must be JSON-serialisable.
        group (str, optional): A heading to group related entries under in the web graph display. Defaults to General.
        data_type (str, optional): How the value is rendered in the web graph. The default 'auto' infers it from the value: pandas DataFrames render as tables, dicts and lists as formatted JSON, Path objects as inline images (image suffixes) or file links, http(s) strings as links, and everything else as plain text. Pass a type explicitly to override the inference: 'image' for a path to an image file inside the node (the value is rewritten relative to the store root), 'link' for a clickable link, 'table' for a pandas DataFrame (stored as columns/rows), 'json' for a dict or list, 'code' for a monospaced snippet (e.g. the SQL or shell command that produced the node), or 'text' for plain text.
        searchable (bool, optional): If True the entry is indexed and can be matched by the store's search methods. Set to False for display-only metadata. Defaults to True.

    Examples:
        >>> with store.create_node(step_type="model", parent=clean_node) as node:
        ...     # Plain values — searchable by default
        ...     node.add_meta("accuracy", 0.94, group="Metrics")
        ...     node.add_meta("epochs", 42, group="Metrics")
        ...     node.add_meta("learning_rate", 1e-3, group="Config")
        ...
        ...     # Dict/list — rendered as formatted JSON in the web graph
        ...     node.add_meta("params", {"optimizer": "adam", "dropout": 0.3}, group="Config")
        ...
        ...     # Table — pandas DataFrame rendered as a sortable table; not searchable
        ...     node.add_meta("results", df, data_type="table", group="Outputs")
        ...
        ...     # Image — path rewritten relative to store root, rendered inline
        ...     fig.savefig(node / "confusion.png")
        ...     node.add_meta("confusion_matrix", node / "confusion.png", group="Figures")
        ...
        ...     # Code — rendered in a monospaced block
        ...     node.add_meta("query", "SELECT * FROM runs WHERE status = 'ok'", data_type="code")
        ...
        ...     # Display-only — visible in the web graph but excluded from find_node
        ...     node.add_meta("notes", "rerun after fixing label encoding", searchable=False)
        ...
        ...     # External link — rendered as a clickable URL
        ...     node.add_meta("wandb_run", "https://wandb.ai/my-org/run/abc123", group="Links")
    """
    if key in _RESERVED_KEYS:
        raise ValueError(
            f"{key!r} is a reserved key set by the store and cannot be "
            "set via add_meta."
        )
    self._set_meta(
        key, value, group=group, data_type=data_type, searchable=searchable
    )

artifacts(contains='*')

Searches this node's directory returning all files excluding internal metadata. Recursively finds all artifacts regardless of storage depth.

Parameters:

Name	Type	Description	Default
contains	str	A glob pattern to filter discovered files. A plain substring also works: it is matched anywhere in the filename, case-insensitively. Defaults to "*" (all files).	'*'

Returns:

Type	Description
List[Path]	List[Path]: The matching file paths, ready to read or pass to a loader directly (no need to prefix the store root). A loose file points inside the node's directory; a packed artifact points at a reassembled copy in the store's read cache. Either way the path holds readable bytes, mirroring what node / "file" returns.

Examples:

>>> node.artifacts("*.csv")
[PosixPath('/data/store/abc12345/sample.csv')]
>>> pd.read_csv(node.artifacts("sample")[0])

Source code in src/ancestree/models.py

def artifacts(self, contains: str = "*") -> List[Path]:
    """
    Searches this node's directory returning all files excluding internal metadata.
    Recursively finds all artifacts regardless of storage depth.

    Args:
        contains (str, optional): A glob pattern to filter discovered files. A plain substring also works: it is matched anywhere in the filename, case-insensitively. Defaults to "*" (all files).

    Returns:
        List[Path]: The matching file paths, ready to read or pass to a
            loader directly (no need to prefix the store root). A loose file
            points inside the node's directory; a packed artifact points at a
            reassembled copy in the store's read cache. Either way the path
            holds readable bytes, mirroring what `node / "file"` returns.

    Examples:
        >>> node.artifacts("*.csv")
        [PosixPath('/data/store/abc12345/sample.csv')]
        >>> pd.read_csv(node.artifacts("sample")[0])
    """
    search_pattern = contains
    if "*" not in contains and "?" not in contains:
        search_pattern = f"*{contains}*"

    artifacts = []
    for rel in sorted(self._artifact_rels()):
        name = rel.rsplit("/", 1)[-1]
        if (
            Path(rel).match(search_pattern)
            or name.lower().find(contains.lower()) != -1
        ):
            artifacts.append(self._resolve(rel))
    return artifacts

__truediv__(relative_loc)

Allows the use of the '/' operator to create paths inside the node's directory, mirroring pathlib. This is the idiomatic way to choose where to write an artifact. Any intermediate directories in the path are created automatically, so the returned path is always ready to write to.

Parameters:

Name	Type	Description	Default
relative_loc	Union[Path, str]	The string or Path object to append to the node's base path.	required

Returns:

Name	Type	Description
Path	Path	The resolved destination inside the node's directory.

Raises:

Type	Description
ValueError	If the path escapes the node's directory.

Examples:

>>> with store.create_node(step_type="clean") as node:
...     df.to_csv(node / "results/cleaned.csv")

Source code in src/ancestree/models.py

def __truediv__(self, relative_loc: Union[Path, str]) -> Path:
    """
    Allows the use of the '/' operator to create paths inside the node's directory, mirroring pathlib.
    This is the idiomatic way to choose where to write an artifact. Any intermediate directories in the path are created automatically, so the returned path is always ready to write to.

    Args:
        relative_loc (Union[Path, str]): The string or Path object to append to the node's base path.

    Returns:
        Path: The resolved destination inside the node's directory.

    Raises:
        ValueError: If the path escapes the node's directory.

    Examples:
        >>> with store.create_node(step_type="clean") as node:
        ...     df.to_csv(node / "results/cleaned.csv")
    """
    target_path = (self.path / relative_loc).resolve()
    if not target_path.is_relative_to(self.path.resolve()):
        raise ValueError(
            f"Artifact path {relative_loc!r} escapes the node directory. "
            "Keep all artifact paths inside the node."
        )
    # A packed artifact with no loose file on disk is a read: reassemble it
    # into the read cache and return that path (reading is transparent). A
    # loose file (read) or a not-yet-written path (write target) returns a
    # path inside the node directory.
    rel = target_path.relative_to(self.path.resolve()).as_posix()
    if not target_path.exists():
        if rel not in self._manifest():
            self._manifest_data = None  # may have been packed in the background
        if rel in self._manifest():
            return self._materialize(rel)
    target_path.parent.mkdir(parents=True, exist_ok=True)
    return target_path