DAG Guide

Overview

TidyRun provides three core building blocks for deferred computation:

• Job: one deferred function call with named kwargs
• ParametrizedJob: a parameter grid over keys that slices into nested jobs
• DAG: a key-addressable mapping of deferred nodes that can be evaluated to disk

A DAG evaluation writes outputs with the same storage contract used by serialize(...), so deserialize(...) returns a matching LazyDict tree.

Core Concepts

Job

A Job holds a Python callable and named arguments.

from tidyrun import Job


def add(x: int, y: int) -> int:
    return x + y


job = Job(func=add, kwargs={"x": 1, "y": 2})

Arguments can be plain values, LazyDict, Job, ParametrizedJob, or DAG.

ParametrizedJob

A ParametrizedJob represents a Cartesian-style parameterized computation. Accessing a key fixes the first parameter:

• returns another ParametrizedJob while parameters remain
• returns a concrete Job when the last parameter is fixed

from tidyrun import ParametrizedJob


def score(model: str, split: str, prefix: str = "") -> str:
    return f"{prefix}{model}:{split}"


grid = ParametrizedJob(
    func=score,
    parameter_names=["model", "split"],
    parameter_values=[("m1", "train"), ("m1", "test"), ("m2", "train")],
    kwargs={"prefix": "run="},
)

model_slice = grid["m1"]
leaf_job = model_slice["train"]

DAG

A DAG maps keys (same key type contract as keys.py) to nodes. Supported node types are:

• Job
• ParametrizedJob
• nested DAG

from tidyrun import DAG, Job


def square(x: int) -> int:
    return x * x


dag = DAG()
dag["a"] = Job(func=square, kwargs={"x": 3})

Evaluation

DAG.evaluate(...) is materialize-first:

1. Compile the DAG into a plan directory (by default <target>/plan)
2. Execute jobs in dependency order
3. Serialize top-level outputs to the outputs directory (by default <target>/outputs)

By default, jobs run in isolated subprocesses.

Core DAG Lifecycle APIs

These three methods cover the most common lifecycle for local and resumable execution:

• evaluate: one-call workflow that materializes a plan, executes it, and writes top-level outputs to your run outputs directory.
• materialize: compile only (no execution). Use this when you want a stable, inspectable on-disk plan before running jobs.
• execute_materialized: run an already materialized plan, optionally with skip_completed=True to resume partially completed runs.

Typical pattern:

1. Use evaluate for everyday runs.
2. Use materialize + execute_materialized for debugging, reproducibility, or resumable workflows.

Default layout for dag.evaluate("./exp1"):

• plan: ./exp1/plan
• outputs: ./exp1/outputs

You can also skip target entirely when both paths are explicit:

result = dag.evaluate(
    dag_path="./exp1-plan",
    output_path="./exp1-outputs",
)

Sequential Evaluation

result = dag.evaluate("./sequential")
print(result["a"])  # 9

Execution Modes

Select execution behavior with execution_mode:

• "subprocess" (default): isolated Python subprocess per job
• "thread": run jobs in the current process (lower overhead)
• "process": run jobs using ProcessPoolExecutor

As a rule of thumb, start with "subprocess" for the safest isolation and reproducibility. Choose "thread" for lightweight local runs where low overhead matters (for example during rapid iteration or tests). Choose "process" when running many local CPU-bound jobs and you want process-level parallelism with worker reuse through a process pool.

# Fast local testing (no subprocess spawn per job)
result = dag.evaluate("./thread-mode", execution_mode="thread")

# Process pool execution
result = dag.evaluate(
    "./process-mode",
    execution_mode="process",
    max_workers=4,
)

Local Parallel Evaluation

Use max_workers to evaluate independent jobs in parallel.

# Thread pool (thread/subprocess modes)
result = dag.evaluate("./threaded", max_workers=4, execution_mode="thread")

# Process pool (process mode)
result = dag.evaluate("./process-pooled", max_workers=4, execution_mode="process")

Failure Handling and Resume

DAG execution fails fast. If any job fails, scheduling stops and a DAGExecutionError is raised with structured context:

• failed_job_id
• cause
• completed_jobs
• cancelled_jobs

from tidyrun import DAGExecutionError

try:
    dag.evaluate("./run")
except DAGExecutionError as exc:
    print("failed:", exc.failed_job_id)
    print("completed:", sorted(exc.completed_jobs))
    print("cancelled:", sorted(exc.cancelled_jobs))

To resume after fixing a failing job, run from an existing materialized plan and set skip_completed=True so already-written outputs are reused:

If outputs already exist and skip_completed=False (default), execute_materialized(...) now raises an error to prevent accidental mixing of previous and newly computed results.

plan_dir = dag.materialize("./run/plan")

result = dag.execute_materialized(
    dag_path=plan_dir,
    output_path="./run/outputs",
    skip_completed=True,
)

Progress Logging

Use progress=True to emit simple progress logs during plan compilation and job execution.

result = dag.evaluate(
    "./run",
    progress=True,
)

You can also provide a custom callback to collect or redirect progress lines:

messages: list[str] = []
result = dag.evaluate(
    "./run",
    progress=True,
    progress_callback=messages.append,
)

If outputs are obsolete or wrong, clear them before resubmitting:

# Remove all outputs
dag.clear_outputs("./run/plan")

# Or remove specific job outputs only
dag.clear_outputs("./run/plan", job_ids=["train/model_a", "metrics/model_a"])

Custom Executor

You can pass your own concurrent.futures.Executor.

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=8) as pool:
    result = dag.evaluate("./custom", executor=pool, execution_mode="thread")

Pass either executor or max_workers, not both.

Materialized Plan Helpers

For debugging/reproducibility, you can inspect materialized artifacts directly:

from tidyrun import load_callable, load_job_definition, load_job_inputs

plan_dir = dag.materialize("./experiment/plan")
definition = load_job_definition(plan_dir, "a")
func = load_callable(definition, plan_dir)
kwargs = load_job_inputs(definition, plan_dir)
print(func(**kwargs))

SLURM Executor

SlurmExecutor submits each job as an sbatch task, polling squeue for completion. The plan directory and shared_dir must both be on shared storage visible from all compute nodes. See the Executors guide for setup instructions, resource configuration, and a full deployment example with git commit pinning.

from tidyrun import SlurmExecutor

with SlurmExecutor(
    shared_dir="/shared/tidyrun_scratch",
    partition="compute",
    time_limit="01:00:00",
    memory="8G",
) as executor:
    result = dag.execute_materialized(
        dag_path="/shared/plans/run-001",
        output_path="/shared/outputs/run-001",
        executor=executor,
    )

AWS Batch Executor

AwsBatchExecutor submits each job as a Batch container task, polling describe_jobs for completion. The plan directory must be an S3 URI, and the container image must call tidyrun-batch-entrypoint as its CMD. See the Executors guide for the container setup, IAM requirements, and a full deployment example with git commit pinning.

from tidyrun import AwsBatchExecutor

with AwsBatchExecutor(
    job_queue="my-queue",
    job_definition="my-worker:1",
) as executor:
    result = dag.execute_materialized(
        dag_path="s3://my-bucket/plans/run-001",
        output_path="s3://my-bucket/outputs/run-001",
        executor=executor,
    )

End-to-End Example

from tidyrun import DAG, Job, ParametrizedJob


def metric(model: str, split: str, base: int = 1) -> str:
    return f"{model}:{split}:{base}"


scores = ParametrizedJob(
    func=metric,
    parameter_names=["model", "split"],
    parameter_values=[("m1", "train"), ("m1", "test"), ("m2", "train")],
    kwargs={"base": 10},
)

summary = Job(func=lambda value: f"summary={value}", kwargs={"value": "ok"})


dag = DAG()
dag["scores"] = scores
dag["summary"] = summary

outputs = dag.evaluate("./experiment", max_workers=4)
print(outputs["scores"]["m1"]["train"])  # m1:train:10
print(outputs["summary"])                 # summary=ok

Notes

• Evaluated outputs are serialized using the same metadata sidecar mechanism as the serialization API.
• The on-disk plan format is intended for reproducibility: you can re-run one job later via run_materialized_job(plan_dir, job_id).
• job_resources is keyed by top-level DAG keys and is only applied when the executor implements submit_with_options(...).

API Reference

tidyrun.job.Job dataclass

A deferred computation: a callable with named arguments.

Each argument value is an Operand, which may be a plain Python value, a LazyDict (existing on-disk outputs), another Job, or a DAG. Arguments are resolved recursively before the function is called.

Example::

def add(x: int, y: int) -> int:
    return x + y

job = Job(func=add, kwargs={"x": 1, "y": 2})

Source code in src/tidyrun/job.py

@dataclass
class Job:
    """A deferred computation: a callable with named arguments.

    Each argument value is an Operand, which may be a plain Python value,
    a LazyDict (existing on-disk outputs), another Job, or a DAG.
    Arguments are resolved recursively before the function is called.

    Example::

        def add(x: int, y: int) -> int:
            return x + y

        job = Job(func=add, kwargs={"x": 1, "y": 2})
    """

    func: Callable[..., Any]
    kwargs: Mapping[str, Operand]

    def __post_init__(self) -> None:
        validate_callable_bindings(
            func=self.func,
            kwargs=self.kwargs,
            parameter_names=(),
        )

    def rerun_snippet(self, *, dag_path: str | Path, job_id: str) -> str:
        """Return a Python snippet that reruns this job from a materialized plan."""
        from tidyrun.plan import rerun_snippet as _rerun_snippet

        return _rerun_snippet(dag_path, job_id)

rerun_snippet(*, dag_path, job_id)

Return a Python snippet that reruns this job from a materialized plan.

Source code in src/tidyrun/job.py

def rerun_snippet(self, *, dag_path: str | Path, job_id: str) -> str:
    """Return a Python snippet that reruns this job from a materialized plan."""
    from tidyrun.plan import rerun_snippet as _rerun_snippet

    return _rerun_snippet(dag_path, job_id)

tidyrun.dag.ParametrizedJob

Bases: DAG

A deferred computation indexed by parameter keys.

Parameters are declared through parameter_names and populated through parameter_values. Accessing a key fixes the first parameter and returns either a :class:Job (when one parameter remains) or another :class:ParametrizedJob (when more parameters remain).

Being a subclass of :class:DAG, a ParametrizedJob inherits all execution methods (materialize, execute_materialized, evaluate, clear_outputs) with identical semantics: the top-level keys are the first-level parameter values and no extra wrapping level is added.

Source code in src/tidyrun/dag.py

class ParametrizedJob(DAG):
    """A deferred computation indexed by parameter keys.

    Parameters are declared through ``parameter_names`` and populated through
    ``parameter_values``. Accessing a key fixes the first parameter and returns
    either a :class:`Job` (when one parameter remains) or another
    :class:`ParametrizedJob` (when more parameters remain).

    Being a subclass of :class:`DAG`, a ``ParametrizedJob`` inherits all
    execution methods (``materialize``, ``execute_materialized``, ``evaluate``,
    ``clear_outputs``) with identical semantics: the top-level keys are the
    first-level parameter values and no extra wrapping level is added.
    """

    func: Callable[..., Any]
    parameter_names: tuple[str, ...]
    parameter_values: tuple[tuple[Key, ...], ...]
    kwargs: Mapping[str, Any]

    def __init__(
        self,
        func: Callable[..., Any],
        parameter_names: list[str] | tuple[str, ...],
        parameter_values: list[tuple[Key, ...]] | tuple[tuple[Key, ...], ...],
        kwargs: Mapping[str, Any] | None = None,
    ) -> None:
        self.func = func
        self.parameter_names = tuple(parameter_names)
        self.parameter_values = tuple(tuple(v) for v in parameter_values)
        self.kwargs = {} if kwargs is None else kwargs
        self._validate()

    @property  # type: ignore[override]
    def _nodes(self) -> dict[Key, Node]:  # pyright: ignore[reportIncompatibleVariableOverride]
        return {k: self[k] for k in self}

    def __setitem__(self, key: Key, value: Node) -> None:
        raise TypeError(f"{type(self).__name__!r} does not support item assignment")

    def __getitem__(self, key: Key) -> Job | ParametrizedJob:
        matching = [values for values in self.parameter_values if values[0] == key]
        if not matching:
            raise KeyError(key)

        parameter_name = self.parameter_names[0]
        bound_kwargs = dict(self.kwargs)
        bound_kwargs[parameter_name] = key

        if len(self.parameter_names) == 1:
            return Job(func=self.func, kwargs=bound_kwargs)
        return ParametrizedJob(
            func=self.func,
            parameter_names=self.parameter_names[1:],
            parameter_values=[values[1:] for values in matching],
            kwargs=bound_kwargs,
        )

    def __iter__(self) -> Iterator[Key]:
        seen: set[Key] = set()
        for values in self.parameter_values:
            first = values[0]
            if first in seen:
                continue
            seen.add(first)
            yield first

    def __len__(self) -> int:
        return len(set(values[0] for values in self.parameter_values))

    def _validate(self) -> None:
        if not self.parameter_names:
            raise ValueError("parameter_names must not be empty")
        if len(set(self.parameter_names)) != len(self.parameter_names):
            raise ValueError("parameter_names must be unique")
        expected_arity = len(self.parameter_names)
        seen: set[tuple[Key, ...]] = set()
        for values in self.parameter_values:
            if len(values) != expected_arity:
                raise ValueError(
                    f"Each parameter tuple must have length {expected_arity}"
                )
            for key in values:
                encode_key(key)
            if values in seen:
                raise ValueError("parameter_values must not contain duplicates")
            seen.add(values)
        validate_callable_bindings(
            func=self.func,
            kwargs=self.kwargs,
            parameter_names=self.parameter_names,
        )

tidyrun.dag.DAG

Bases: Mapping[Key, Node]

A mapping from keys to deferred computations.

DAG is the write-time dual of LazyDict: it maps :data:Key values to nodes (:class:~tidyrun.Job, :class:~tidyrun.ParametrizedJob, or nested :class:DAG instances). Evaluating a DAG to disk produces the same on-disk layout as :func:~tidyrun.serialize, so that :func:~tidyrun.deserialize returns a :class:~tidyrun.LazyDict with the same key tree.

DAG execution is materialized-first: compile a plan on disk, then execute each job in a dedicated Python subprocess with arguments loaded through :func:~tidyrun.deserialize.

Example::

import tempfile, pathlib
from tidyrun.job import Job
from tidyrun.dag import DAG

def square(x: int) -> int:
    return x * x

dag = DAG()
dag["a"] = Job(func=square, kwargs={"x": 3})

with tempfile.TemporaryDirectory() as tmp:
    result = dag.evaluate(pathlib.Path(tmp) / "outputs")
    assert result["a"] == 9

Source code in src/tidyrun/dag.py

class DAG(Mapping[Key, Node]):
    """A mapping from keys to deferred computations.

    DAG is the write-time dual of LazyDict: it maps :data:`Key` values to
    nodes (:class:`~tidyrun.Job`, :class:`~tidyrun.ParametrizedJob`, or nested
    :class:`DAG` instances). Evaluating a DAG to disk produces the same on-disk
    layout as
    :func:`~tidyrun.serialize`, so that :func:`~tidyrun.deserialize` returns a
    :class:`~tidyrun.LazyDict` with the same key tree.

    DAG execution is materialized-first: compile a plan on disk, then execute
    each job in a dedicated Python subprocess with arguments loaded through
    :func:`~tidyrun.deserialize`.

    Example::

        import tempfile, pathlib
        from tidyrun.job import Job
        from tidyrun.dag import DAG

        def square(x: int) -> int:
            return x * x

        dag = DAG()
        dag["a"] = Job(func=square, kwargs={"x": 3})

        with tempfile.TemporaryDirectory() as tmp:
            result = dag.evaluate(pathlib.Path(tmp) / "outputs")
            assert result["a"] == 9
    """

    def __init__(self, nodes: Mapping[Key, Node] | None = None) -> None:
        self._nodes: dict[Key, Node] = dict(nodes) if nodes is not None else {}

    def __getitem__(self, key: Key) -> Node:
        return self._nodes[key]

    def __setitem__(self, key: Key, value: Node) -> None:
        self._nodes[key] = value

    def __iter__(self) -> Iterator[Key]:
        return iter(self._nodes)

    def __len__(self) -> int:
        return len(self._nodes)

    def materialize(
        self,
        dag_path: Any,
        *,
        prefix: str | None = None,
        progress: bool = False,
        progress_callback: ProgressCallback | None = None,
    ) -> Path | CloudPath:
        """Write job definitions and literal inputs for process execution.

        Parameters
        ----------
        dag_path :
            Destination for the materialized plan. May be a plain path
            (definitions, inputs, and outputs are created as subdirectories)
            or a :class:`PlanPaths` object that places each component at an
            independent location. Accepts ``s3://`` URIs when the optional
            ``boto3`` dependency is installed.
        prefix :
            Optional string prepended to all job IDs in this plan.
        progress :
            When ``True``, emit progress logs during compilation.
        progress_callback :
            Optional callback that receives each progress message string.

        Returns
        -------
        Path
            The plan root directory (``dag_path`` as a ``Path``, or the
            first component's parent for a ``PlanPaths``).
        """
        seen_nodes: set[int] = set()
        total_jobs = sum(
            _count_unique_jobs(node, seen_nodes) for node in self._nodes.values()
        )
        reporter = _ProgressReporter(
            enabled=progress,
            callback=progress_callback,
            phase="materialize",
            total=total_jobs,
        )
        reporter.info(f"starting ({total_jobs} jobs)")

        if isinstance(dag_path, PlanPaths):
            _PlanCompiler(dag_path, reporter).compile(self._nodes, prefix)
            reporter.info("done")
            return dag_path.definitions.parent

        if is_s3_location(dag_path):
            # Compile locally, then upload the whole plan tree.
            with TemporaryDirectory() as temp_root:
                plan_dir = Path(temp_root) / _s3_leaf_name(dag_path)
                _PlanCompiler(PlanPaths.from_root(plan_dir), reporter).compile(
                    self._nodes, prefix
                )
                upload_local_tree_to_s3(plan_dir.parent, dag_path)
            reporter.info("done")
            return AnyPath(dag_path)

        plan_dir = to_path(dag_path)
        _PlanCompiler(PlanPaths.from_root(plan_dir), reporter).compile(
            self._nodes, prefix
        )
        reporter.info("done")
        return plan_dir

    def execute_materialized(
        self,
        dag_path: Any,
        executor: Executor | None = None,
        max_workers: int | None = None,
        job_resources: Mapping[Key, Mapping[str, str | int]] | None = None,
        execution_mode: ExecutionMode = "subprocess",
        skip_completed: bool = False,
        skip_running: bool = False,
        progress: bool = False,
        progress_callback: ProgressCallback | None = None,
    ) -> Any:
        """Execute a previously materialized plan with dependency ordering.

        Parameters
        ----------
        dag_path:
            Path to the materialized DAG directory. Outputs are always written
            to ``dag_path/outputs``.
        executor:
            Optional custom :class:`~concurrent.futures.Executor`.
        max_workers:
            Number of workers for parallel execution. Creates a
            ThreadPoolExecutor if execution_mode is "thread", or
            ProcessPoolExecutor if execution_mode is "process".
        job_resources:
            Optional per-node submission options.
        execution_mode:
            How to execute jobs: "subprocess" (default, isolated Python processes),
            "thread" (shared memory in threads), or "process" (ProcessPoolExecutor).
        skip_completed:
            When ``True``, skip any job whose output already exists on disk.
            This enables resuming a partially-completed DAG after a failure
            without re-running jobs that already succeeded.
        skip_running:
            When ``True``, skip jobs whose ``.running`` sentinel exists.
        progress:
            When ``True``, emit progress logs while executing jobs.
        progress_callback:
            Optional callback used for progress messages.
        """
        from tidyrun.serialization.api import deserialize
        from tidyrun.serialization.metadata import metadata_exists

        if executor is not None and max_workers is not None:
            raise ValueError("Pass either executor or max_workers, not both")

        plan_dir = to_path(dag_path)
        plan_paths = PlanPaths.from_root(plan_dir)
        plan_paths.outputs.mkdir(parents=True, exist_ok=True)

        if not plan_paths.definitions.is_dir():
            raise ValueError(
                f"No materialized plan found at {plan_dir}. Run materialize() first."
            )
        graph = read_plan_graph(plan_paths.definitions)

        # Synthetic aggregator jobs for every group node in the DAG tree. They
        # run inline (no subprocess) after their children complete and write
        # the dict-folder .tidyrun metadata for intermediate output folders.
        aggregator_deps: dict[str, list[str]] = {}
        root_children = [
            _build_aggregator_deps(node, _encode_key_checked(key), aggregator_deps)
            for key, node in self._nodes.items()
        ]
        dependencies = dict(graph.dependencies)
        for agg_id, child_ids in aggregator_deps.items():
            dependencies[agg_id] = set(child_ids)
        inline_runners: dict[str, Callable[[], None]] = {
            agg_id: partial(write_group_metadata, agg_id, child_ids, plan_paths.outputs)
            for agg_id, child_ids in aggregator_deps.items()
        }

        resources_by_key: Mapping[Key, Mapping[str, str | int]] = (
            {} if job_resources is None else job_resources
        )
        unknown_keys = [key for key in resources_by_key if key not in self._nodes]
        if unknown_keys:
            raise ValueError(f"job_resources contains unknown DAG keys: {unknown_keys}")
        # Per-job submission options currently apply to top-level Job nodes only.
        resources_by_job_id = {
            _encode_key_checked(key): dict(options)
            for key, options in resources_by_key.items()
            if isinstance(self._nodes[key], Job)
        }

        reporter = _ProgressReporter(
            enabled=progress,
            callback=progress_callback,
            phase="execute",
            total=len(graph.dependencies),
        )
        reporter.info(f"starting ({len(graph.dependencies)} jobs)")

        # Guard against accidentally mixing old and new results when reusing a
        # partially executed plan without resume semantics.
        if not skip_completed:
            existing_outputs = sorted(
                job_id
                for job_id in graph.dependencies
                if job_output_exists(plan_paths.outputs, job_id)
            )
            if existing_outputs:
                raise ValueError(
                    "Materialized plan already has existing job outputs. "
                    "Use skip_completed=True to resume, or clear outputs before "
                    "re-running. Existing job ids: "
                    f"{existing_outputs}"
                )

        execute_graph(
            dependencies,
            plan_paths,
            plan_dir,
            executor=executor,
            max_workers=max_workers,
            execution_mode=execution_mode,
            skip_completed=skip_completed,
            skip_running=skip_running,
            reporter=reporter,
            inline_runners=inline_runners,
            array_groups=graph.array_groups,
            array_group_by_job_id=graph.array_group_by_job_id,
            resources_by_job_id=resources_by_job_id,
        )

        # Write the root dict-folder .tidyrun for the outputs directory so the
        # on-disk layout is identical to serialize(dict, outputs_path).
        if not (skip_completed and metadata_exists(plan_paths.outputs)):
            write_root_metadata(plan_paths.outputs, root_children)

        reporter.info("done")
        return deserialize(plan_paths.outputs)

    def evaluate(
        self,
        dag_path: Any,
        executor: Executor | None = None,
        max_workers: int | None = None,
        job_resources: Mapping[Key, Mapping[str, str | int]] | None = None,
        execution_mode: ExecutionMode = "subprocess",
        skip_completed: bool = False,
        progress: bool = False,
        progress_callback: ProgressCallback | None = None,
    ) -> Any:
        """Evaluate this DAG to disk (materialize, then execute).

        Parameters
        ----------
        dag_path:
            Directory for the materialized plan. Outputs are written to
            ``dag_path/outputs``.
        executor:
            Optional :class:`~concurrent.futures.Executor` for parallel
            job launches.
        max_workers:
            Number of local workers for parallel evaluation. When set,
            creates a :class:`~concurrent.futures.ThreadPoolExecutor` (for
            "thread" or "subprocess" modes) or
            :class:`~concurrent.futures.ProcessPoolExecutor` (for "process" mode).
            Cannot be combined with `executor`.
        job_resources:
            Optional per-node submission options keyed by DAG key. This is
            primarily useful with executors that expose
            ``submit_with_options(..., sbatch_options=...)``, such as
            ``SlurmExecutor``.
        execution_mode:
            How to execute jobs:

            - ``"subprocess"`` (default): Each job runs in an isolated Python
              subprocess with full process separation. Safest for reproducibility
              but has subprocess spawn overhead.
            - ``"thread"``: Jobs run in threads within the same Python process
              with shared memory. Fast for test DAGs with small jobs but may
              encounter GIL contention.
            - ``"process"``: Jobs run in separate processes via
              :class:`~concurrent.futures.ProcessPoolExecutor`. Similar to
              subprocess but with potentially faster worker pool management.
        skip_completed:
            When ``True``, skip jobs whose outputs already exist in the
            materialized plan.
        progress:
            When ``True``, emit progress logs for materialization and execution.
        progress_callback:
            Optional callback used for progress messages.

        Returns
        -------
        LazyDict
            The deserialized :class:`~tidyrun.LazyDict` at ``dag_path/outputs``
            after all nodes have been written.
        """
        plan_dir = self.materialize(
            dag_path,
            progress=progress,
            progress_callback=progress_callback,
        )
        return self.execute_materialized(
            plan_dir,
            executor=executor,
            max_workers=max_workers,
            job_resources=job_resources,
            execution_mode=execution_mode,
            skip_completed=skip_completed,
            progress=progress,
            progress_callback=progress_callback,
        )

    def evaluate_in_subprocesses(
        self,
        dag_path: Any,
        executor: Executor | None = None,
        max_workers: int | None = None,
        job_resources: Mapping[Key, Mapping[str, str | int]] | None = None,
        execution_mode: ExecutionMode = "subprocess",
        skip_completed: bool = False,
        progress: bool = False,
        progress_callback: ProgressCallback | None = None,
    ) -> Any:
        """Alias of :meth:`evaluate`, kept for backward compatibility."""
        return self.evaluate(
            dag_path,
            executor=executor,
            max_workers=max_workers,
            job_resources=job_resources,
            execution_mode=execution_mode,
            skip_completed=skip_completed,
            progress=progress,
            progress_callback=progress_callback,
        )

    def clear_outputs(
        self,
        dag_path: Any,
        job_ids: list[str] | None = None,
    ) -> None:
        """Delete serialized outputs for jobs in a materialized plan.

        Use this to discard stale or incorrect outputs before resubmitting a
        DAG.  When *job_ids* is ``None`` the entire outputs directory is
        removed.  Otherwise only the specified jobs' output files are deleted.

        Parameters
        ----------
        dag_path:
            Path to the materialized DAG directory.
        job_ids:
            Optional list of job IDs whose outputs should be cleared.
            When ``None``, all outputs are removed.
        """
        import shutil

        from tidyrun.serialization.metadata import metadata_path, read_metadata

        plan_dir = to_path(dag_path)
        outputs_dir = PlanPaths.from_root(plan_dir).outputs

        if job_ids is None:
            if outputs_dir.exists():
                shutil.rmtree(outputs_dir)
            return

        for job_id in job_ids:
            base = job_output_base(outputs_dir, job_id)
            meta = metadata_path(base)
            if not meta.is_file():
                continue
            try:
                suffix = read_metadata(base).get("suffix", "")
            except Exception:
                # A corrupt metadata file should not prevent clearing the job;
                # fall back to deleting the suffix-less payload path.
                suffix = ""
            payload = Path(str(base) + suffix) if suffix else base
            if payload.exists():
                if payload.is_dir():
                    shutil.rmtree(payload)
                else:
                    payload.unlink()
            meta.unlink()

materialize(dag_path, *, prefix=None, progress=False, progress_callback=None)

Write job definitions and literal inputs for process execution.

Parameters:

Name	Type	Description	Default
dag_path	Any	Destination for the materialized plan. May be a plain path (definitions, inputs, and outputs are created as subdirectories) or a :class:PlanPaths object that places each component at an independent location. Accepts s3:// URIs when the optional boto3 dependency is installed.	required
prefix	str | None	Optional string prepended to all job IDs in this plan.	None
progress	bool	When True, emit progress logs during compilation.	False
progress_callback	ProgressCallback | None	Optional callback that receives each progress message string.	None

Returns:

Type	Description
Path	The plan root directory (dag_path as a Path, or the first component's parent for a PlanPaths).

Source code in src/tidyrun/dag.py

def materialize(
    self,
    dag_path: Any,
    *,
    prefix: str | None = None,
    progress: bool = False,
    progress_callback: ProgressCallback | None = None,
) -> Path | CloudPath:
    """Write job definitions and literal inputs for process execution.

    Parameters
    ----------
    dag_path :
        Destination for the materialized plan. May be a plain path
        (definitions, inputs, and outputs are created as subdirectories)
        or a :class:`PlanPaths` object that places each component at an
        independent location. Accepts ``s3://`` URIs when the optional
        ``boto3`` dependency is installed.
    prefix :
        Optional string prepended to all job IDs in this plan.
    progress :
        When ``True``, emit progress logs during compilation.
    progress_callback :
        Optional callback that receives each progress message string.

    Returns
    -------
    Path
        The plan root directory (``dag_path`` as a ``Path``, or the
        first component's parent for a ``PlanPaths``).
    """
    seen_nodes: set[int] = set()
    total_jobs = sum(
        _count_unique_jobs(node, seen_nodes) for node in self._nodes.values()
    )
    reporter = _ProgressReporter(
        enabled=progress,
        callback=progress_callback,
        phase="materialize",
        total=total_jobs,
    )
    reporter.info(f"starting ({total_jobs} jobs)")

    if isinstance(dag_path, PlanPaths):
        _PlanCompiler(dag_path, reporter).compile(self._nodes, prefix)
        reporter.info("done")
        return dag_path.definitions.parent

    if is_s3_location(dag_path):
        # Compile locally, then upload the whole plan tree.
        with TemporaryDirectory() as temp_root:
            plan_dir = Path(temp_root) / _s3_leaf_name(dag_path)
            _PlanCompiler(PlanPaths.from_root(plan_dir), reporter).compile(
                self._nodes, prefix
            )
            upload_local_tree_to_s3(plan_dir.parent, dag_path)
        reporter.info("done")
        return AnyPath(dag_path)

    plan_dir = to_path(dag_path)
    _PlanCompiler(PlanPaths.from_root(plan_dir), reporter).compile(
        self._nodes, prefix
    )
    reporter.info("done")
    return plan_dir

execute_materialized(dag_path, executor=None, max_workers=None, job_resources=None, execution_mode='subprocess', skip_completed=False, skip_running=False, progress=False, progress_callback=None)

Execute a previously materialized plan with dependency ordering.

Parameters:

Name	Type	Description	Default
dag_path	Any	Path to the materialized DAG directory. Outputs are always written to dag_path/outputs.	required
executor	Executor | None	Optional custom :class:~concurrent.futures.Executor.	None
max_workers	int | None	Number of workers for parallel execution. Creates a ThreadPoolExecutor if execution_mode is "thread", or ProcessPoolExecutor if execution_mode is "process".	None
job_resources	Mapping[Key, Mapping[str, str | int]] | None	Optional per-node submission options.	None
execution_mode	ExecutionMode	How to execute jobs: "subprocess" (default, isolated Python processes), "thread" (shared memory in threads), or "process" (ProcessPoolExecutor).	'subprocess'
skip_completed	bool	When True, skip any job whose output already exists on disk. This enables resuming a partially-completed DAG after a failure without re-running jobs that already succeeded.	False
skip_running	bool	When True, skip jobs whose .running sentinel exists.	False
progress	bool	When True, emit progress logs while executing jobs.	False
progress_callback	ProgressCallback | None	Optional callback used for progress messages.	None

Source code in src/tidyrun/dag.py

def execute_materialized(
    self,
    dag_path: Any,
    executor: Executor | None = None,
    max_workers: int | None = None,
    job_resources: Mapping[Key, Mapping[str, str | int]] | None = None,
    execution_mode: ExecutionMode = "subprocess",
    skip_completed: bool = False,
    skip_running: bool = False,
    progress: bool = False,
    progress_callback: ProgressCallback | None = None,
) -> Any:
    """Execute a previously materialized plan with dependency ordering.

    Parameters
    ----------
    dag_path:
        Path to the materialized DAG directory. Outputs are always written
        to ``dag_path/outputs``.
    executor:
        Optional custom :class:`~concurrent.futures.Executor`.
    max_workers:
        Number of workers for parallel execution. Creates a
        ThreadPoolExecutor if execution_mode is "thread", or
        ProcessPoolExecutor if execution_mode is "process".
    job_resources:
        Optional per-node submission options.
    execution_mode:
        How to execute jobs: "subprocess" (default, isolated Python processes),
        "thread" (shared memory in threads), or "process" (ProcessPoolExecutor).
    skip_completed:
        When ``True``, skip any job whose output already exists on disk.
        This enables resuming a partially-completed DAG after a failure
        without re-running jobs that already succeeded.
    skip_running:
        When ``True``, skip jobs whose ``.running`` sentinel exists.
    progress:
        When ``True``, emit progress logs while executing jobs.
    progress_callback:
        Optional callback used for progress messages.
    """
    from tidyrun.serialization.api import deserialize
    from tidyrun.serialization.metadata import metadata_exists

    if executor is not None and max_workers is not None:
        raise ValueError("Pass either executor or max_workers, not both")

    plan_dir = to_path(dag_path)
    plan_paths = PlanPaths.from_root(plan_dir)
    plan_paths.outputs.mkdir(parents=True, exist_ok=True)

    if not plan_paths.definitions.is_dir():
        raise ValueError(
            f"No materialized plan found at {plan_dir}. Run materialize() first."
        )
    graph = read_plan_graph(plan_paths.definitions)

    # Synthetic aggregator jobs for every group node in the DAG tree. They
    # run inline (no subprocess) after their children complete and write
    # the dict-folder .tidyrun metadata for intermediate output folders.
    aggregator_deps: dict[str, list[str]] = {}
    root_children = [
        _build_aggregator_deps(node, _encode_key_checked(key), aggregator_deps)
        for key, node in self._nodes.items()
    ]
    dependencies = dict(graph.dependencies)
    for agg_id, child_ids in aggregator_deps.items():
        dependencies[agg_id] = set(child_ids)
    inline_runners: dict[str, Callable[[], None]] = {
        agg_id: partial(write_group_metadata, agg_id, child_ids, plan_paths.outputs)
        for agg_id, child_ids in aggregator_deps.items()
    }

    resources_by_key: Mapping[Key, Mapping[str, str | int]] = (
        {} if job_resources is None else job_resources
    )
    unknown_keys = [key for key in resources_by_key if key not in self._nodes]
    if unknown_keys:
        raise ValueError(f"job_resources contains unknown DAG keys: {unknown_keys}")
    # Per-job submission options currently apply to top-level Job nodes only.
    resources_by_job_id = {
        _encode_key_checked(key): dict(options)
        for key, options in resources_by_key.items()
        if isinstance(self._nodes[key], Job)
    }

    reporter = _ProgressReporter(
        enabled=progress,
        callback=progress_callback,
        phase="execute",
        total=len(graph.dependencies),
    )
    reporter.info(f"starting ({len(graph.dependencies)} jobs)")

    # Guard against accidentally mixing old and new results when reusing a
    # partially executed plan without resume semantics.
    if not skip_completed:
        existing_outputs = sorted(
            job_id
            for job_id in graph.dependencies
            if job_output_exists(plan_paths.outputs, job_id)
        )
        if existing_outputs:
            raise ValueError(
                "Materialized plan already has existing job outputs. "
                "Use skip_completed=True to resume, or clear outputs before "
                "re-running. Existing job ids: "
                f"{existing_outputs}"
            )

    execute_graph(
        dependencies,
        plan_paths,
        plan_dir,
        executor=executor,
        max_workers=max_workers,
        execution_mode=execution_mode,
        skip_completed=skip_completed,
        skip_running=skip_running,
        reporter=reporter,
        inline_runners=inline_runners,
        array_groups=graph.array_groups,
        array_group_by_job_id=graph.array_group_by_job_id,
        resources_by_job_id=resources_by_job_id,
    )

    # Write the root dict-folder .tidyrun for the outputs directory so the
    # on-disk layout is identical to serialize(dict, outputs_path).
    if not (skip_completed and metadata_exists(plan_paths.outputs)):
        write_root_metadata(plan_paths.outputs, root_children)

    reporter.info("done")
    return deserialize(plan_paths.outputs)

evaluate(dag_path, executor=None, max_workers=None, job_resources=None, execution_mode='subprocess', skip_completed=False, progress=False, progress_callback=None)

Evaluate this DAG to disk (materialize, then execute).

Parameters:

Name	Type	Description	Default
dag_path	Any	Directory for the materialized plan. Outputs are written to dag_path/outputs.	required
executor	Executor | None	Optional :class:~concurrent.futures.Executor for parallel job launches.	None
max_workers	int | None	Number of local workers for parallel evaluation. When set, creates a :class:~concurrent.futures.ThreadPoolExecutor (for "thread" or "subprocess" modes) or :class:~concurrent.futures.ProcessPoolExecutor (for "process" mode). Cannot be combined with executor.	None
job_resources	Mapping[Key, Mapping[str, str | int]] | None	Optional per-node submission options keyed by DAG key. This is primarily useful with executors that expose submit_with_options(..., sbatch_options=...), such as SlurmExecutor.	None
execution_mode	ExecutionMode	How to execute jobs: "subprocess" (default): Each job runs in an isolated Python subprocess with full process separation. Safest for reproducibility but has subprocess spawn overhead. "thread": Jobs run in threads within the same Python process with shared memory. Fast for test DAGs with small jobs but may encounter GIL contention. "process": Jobs run in separate processes via :class:~concurrent.futures.ProcessPoolExecutor. Similar to subprocess but with potentially faster worker pool management.	'subprocess'
skip_completed	bool	When True, skip jobs whose outputs already exist in the materialized plan.	False
progress	bool	When True, emit progress logs for materialization and execution.	False
progress_callback	ProgressCallback | None	Optional callback used for progress messages.	None

Returns:

Type	Description
LazyDict	The deserialized :class:~tidyrun.LazyDict at dag_path/outputs after all nodes have been written.

Source code in src/tidyrun/dag.py

def evaluate(
    self,
    dag_path: Any,
    executor: Executor | None = None,
    max_workers: int | None = None,
    job_resources: Mapping[Key, Mapping[str, str | int]] | None = None,
    execution_mode: ExecutionMode = "subprocess",
    skip_completed: bool = False,
    progress: bool = False,
    progress_callback: ProgressCallback | None = None,
) -> Any:
    """Evaluate this DAG to disk (materialize, then execute).

    Parameters
    ----------
    dag_path:
        Directory for the materialized plan. Outputs are written to
        ``dag_path/outputs``.
    executor:
        Optional :class:`~concurrent.futures.Executor` for parallel
        job launches.
    max_workers:
        Number of local workers for parallel evaluation. When set,
        creates a :class:`~concurrent.futures.ThreadPoolExecutor` (for
        "thread" or "subprocess" modes) or
        :class:`~concurrent.futures.ProcessPoolExecutor` (for "process" mode).
        Cannot be combined with `executor`.
    job_resources:
        Optional per-node submission options keyed by DAG key. This is
        primarily useful with executors that expose
        ``submit_with_options(..., sbatch_options=...)``, such as
        ``SlurmExecutor``.
    execution_mode:
        How to execute jobs:

        - ``"subprocess"`` (default): Each job runs in an isolated Python
          subprocess with full process separation. Safest for reproducibility
          but has subprocess spawn overhead.
        - ``"thread"``: Jobs run in threads within the same Python process
          with shared memory. Fast for test DAGs with small jobs but may
          encounter GIL contention.
        - ``"process"``: Jobs run in separate processes via
          :class:`~concurrent.futures.ProcessPoolExecutor`. Similar to
          subprocess but with potentially faster worker pool management.
    skip_completed:
        When ``True``, skip jobs whose outputs already exist in the
        materialized plan.
    progress:
        When ``True``, emit progress logs for materialization and execution.
    progress_callback:
        Optional callback used for progress messages.

    Returns
    -------
    LazyDict
        The deserialized :class:`~tidyrun.LazyDict` at ``dag_path/outputs``
        after all nodes have been written.
    """
    plan_dir = self.materialize(
        dag_path,
        progress=progress,
        progress_callback=progress_callback,
    )
    return self.execute_materialized(
        plan_dir,
        executor=executor,
        max_workers=max_workers,
        job_resources=job_resources,
        execution_mode=execution_mode,
        skip_completed=skip_completed,
        progress=progress,
        progress_callback=progress_callback,
    )

clear_outputs(dag_path, job_ids=None)

Delete serialized outputs for jobs in a materialized plan.

Use this to discard stale or incorrect outputs before resubmitting a DAG. When job_ids is None the entire outputs directory is removed. Otherwise only the specified jobs' output files are deleted.

Parameters:

Name	Type	Description	Default
dag_path	Any	Path to the materialized DAG directory.	required
job_ids	list[str] | None	Optional list of job IDs whose outputs should be cleared. When None, all outputs are removed.	None

Source code in src/tidyrun/dag.py

def clear_outputs(
    self,
    dag_path: Any,
    job_ids: list[str] | None = None,
) -> None:
    """Delete serialized outputs for jobs in a materialized plan.

    Use this to discard stale or incorrect outputs before resubmitting a
    DAG.  When *job_ids* is ``None`` the entire outputs directory is
    removed.  Otherwise only the specified jobs' output files are deleted.

    Parameters
    ----------
    dag_path:
        Path to the materialized DAG directory.
    job_ids:
        Optional list of job IDs whose outputs should be cleared.
        When ``None``, all outputs are removed.
    """
    import shutil

    from tidyrun.serialization.metadata import metadata_path, read_metadata

    plan_dir = to_path(dag_path)
    outputs_dir = PlanPaths.from_root(plan_dir).outputs

    if job_ids is None:
        if outputs_dir.exists():
            shutil.rmtree(outputs_dir)
        return

    for job_id in job_ids:
        base = job_output_base(outputs_dir, job_id)
        meta = metadata_path(base)
        if not meta.is_file():
            continue
        try:
            suffix = read_metadata(base).get("suffix", "")
        except Exception:
            # A corrupt metadata file should not prevent clearing the job;
            # fall back to deleting the suffix-less payload path.
            suffix = ""
        payload = Path(str(base) + suffix) if suffix else base
        if payload.exists():
            if payload.is_dir():
                shutil.rmtree(payload)
            else:
                payload.unlink()
        meta.unlink()