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:

Compile the DAG into a plan directory (by default <target>/plan)
Execute jobs in dependency order
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:

Use evaluate for everyday runs.
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",  # outputs go to dag_path/outputs
        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 — running the same tidyrun version as the submitting machine. 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",  # outputs go to dag_path/outputs
        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]
    #: Provenance ``(parent, key)`` set by ``ParametrizedJob.__getitem__`` when
    #: this job is a subset of a parametrized job. The plan compiler uses it to
    #: resolve dependencies on subsets of DAG members, whose object identity
    #: differs from the registered member's.
    _derived_from: tuple[Any, Any] | None = field(
        default=None, init=False, repr=False, compare=False
    )

    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]
    #: Provenance ``(parent, key)`` when this node is a subset of another
    #: ParametrizedJob; see :attr:`tidyrun.job.Job._derived_from`.
    _derived_from: tuple["ParametrizedJob", Key] | None = None

    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

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

    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, CloudPath):
            # Normalize cloud path objects to their URI so the S3 branch below
            # handles them like s3:// strings.
            dag_path = str(dag_path)

        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. Dependency
            # links are written as sidecar files: local symlinks would not
            # survive the upload to S3.
            with TemporaryDirectory() as temp_root:
                plan_dir = Path(temp_root) / _s3_leaf_name(dag_path)
                _PlanCompiler(
                    PlanPaths.from_root(plan_dir),
                    reporter,
                    use_sidecar_dep_links=True,
                ).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
        from tidyrun.serialization.paths import with_suffix

        def _remove_tree(path: Path | CloudPath) -> None:
            if isinstance(path, CloudPath):
                path.rmtree()
            else:
                shutil.rmtree(path)

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

        if job_ids is None:
            if outputs_dir.exists():
                _remove_tree(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 = with_suffix(base, suffix) if suffix else base
            if payload.exists():
                if payload.is_dir():
                    _remove_tree(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, CloudPath):
        # Normalize cloud path objects to their URI so the S3 branch below
        # handles them like s3:// strings.
        dag_path = str(dag_path)

    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. Dependency
        # links are written as sidecar files: local symlinks would not
        # survive the upload to S3.
        with TemporaryDirectory() as temp_root:
            plan_dir = Path(temp_root) / _s3_leaf_name(dag_path)
            _PlanCompiler(
                PlanPaths.from_root(plan_dir),
                reporter,
                use_sidecar_dep_links=True,
            ).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
    from tidyrun.serialization.paths import with_suffix

    def _remove_tree(path: Path | CloudPath) -> None:
        if isinstance(path, CloudPath):
            path.rmtree()
        else:
            shutil.rmtree(path)

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

    if job_ids is None:
        if outputs_dir.exists():
            _remove_tree(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 = with_suffix(base, suffix) if suffix else base
        if payload.exists():
            if payload.is_dir():
                _remove_tree(payload)
            else:
                payload.unlink()
        meta.unlink()

DAG Guide¶

Overview¶

Core Concepts¶

Job¶

ParametrizedJob¶

DAG¶

Evaluation¶

Core DAG Lifecycle APIs¶

Sequential Evaluation¶

Execution Modes¶

Local Parallel Evaluation¶

Failure Handling and Resume¶

Progress Logging¶

Custom Executor¶

Materialized Plan Helpers¶

SLURM Executor¶

AWS Batch Executor¶

End-to-End Example¶

Notes¶

API Reference¶

tidyrun.job.Job dataclass ¶

rerun_snippet(*, dag_path, job_id) ¶

tidyrun.dag.ParametrizedJob ¶

tidyrun.dag.DAG ¶

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

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) ¶

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

clear_outputs(dag_path, job_ids=None) ¶

`tidyrun.job.Job` `dataclass` ¶

`rerun_snippet(*, dag_path, job_id)` ¶

`tidyrun.dag.ParametrizedJob` ¶

`tidyrun.dag.DAG` ¶

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

`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)` ¶

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

`clear_outputs(dag_path, job_ids=None)` ¶