Usage Guide¶

Use this page as a workflow-pattern chooser. It is intentionally compact: pick the pattern that matches your step shape, then follow the linked page when you need deeper API or integration detail.

If you are new to Consist, start with the quickstart and first workflow tutorials first.

Choose a Pattern¶

For cache semantics, hydration policies, and cache-miss diagnosis, see Caching and Hydration.

Tracker.run(...): Single Cached Step¶

Use Tracker.run(...) when a step can be represented as a normal Python function. Declare inputs that affect identity, declare expected outputs, and let Consist decide whether to execute or return cached outputs.

from pathlib import Path

import pandas as pd

from consist import ExecutionOptions, Tracker

tracker = Tracker(run_dir="./runs", db_path="./provenance.duckdb")


def summarize_trips(trips_path: Path, threshold: float) -> dict[str, Path]:
    trips = pd.read_parquet(trips_path)
    out = Path("./summary.parquet")
    trips.loc[trips["distance"] >= threshold].groupby("mode").size().to_frame(
        "count"
    ).to_parquet(out)
    return {"summary": out}


result = tracker.run(
    name="summarize_trips",
    fn=summarize_trips,
    inputs={"trips_path": Path("inputs/trips.parquet")},
    config={"threshold": 1.0},
    outputs=["summary"],
    execution_options=ExecutionOptions(input_binding="paths"),
)

summary_artifact = result.outputs["summary"]

Reach for this first when:

• The step has a clear callable boundary.
• Inputs and config fully describe the computation.
• Outputs can be returned as dict[str, Path] or declared with output_paths={...}.
• Use ArtifactSpec(path=..., schema=..., profile_file_schema=True) inside output_paths when a legacy file-writing step needs non-strict logical schema metadata at the same declaration site as the output path.

Use options objects for policy controls:

from consist import CacheOptions, ExecutionOptions, OutputPolicyOptions

result = tracker.run(
    fn=summarize_trips,
    inputs={"trips_path": Path("inputs/trips.parquet")},
    outputs=["summary"],
    cache_options=CacheOptions(
        cache_mode="reuse",
        cache_hydration="inputs-missing",
        code_identity="callable_module",
    ),
    output_policy=OutputPolicyOptions(output_missing="error"),
    execution_options=ExecutionOptions(input_binding="paths"),
)

For the full run API contract, see Run.

Input Binding¶

input_binding controls what Consist passes into your callable for named inputs={...}. It is a runtime execution choice, not the whole signature model.

Prefer "paths" for production pipelines because the I/O boundary stays visible. Use "loaded" when reducing boilerplate matters more than showing the file boundary in the function signature.

runtime_kwargs is still useful for values that should be passed to the callable but should not be treated as declared inputs.

Path Staging Example¶

Requested input staging is for path-bound tools that need a declared input at a specific local destination. inputs={...} remains the identity and lineage source; input_paths={...} describes the workspace location the callable should see.

from pathlib import Path
import subprocess

from consist import ExecutionOptions, Tracker

tracker = Tracker(run_dir="./runs", db_path="./provenance.duckdb")


def run_legacy_tool(config_path: Path) -> dict[str, Path]:
    report = Path("./report.txt")
    subprocess.run(
        ["legacy-tool", "--config", str(config_path), "--out", str(report)],
        check=True,
    )
    return {"report": report}


result = tracker.run(
    name="legacy_report",
    fn=run_legacy_tool,
    inputs={"config_path": Path("./configs/baseline.yaml")},
    outputs=["report"],
    execution_options=ExecutionOptions(
        input_binding="paths",
        input_materialization="requested",
        input_materialization_mode="copy",
        input_paths={
            "config_path": tracker.run_dir / "workspace" / "tool-config.yaml"
        },
    ),
)

Use this when:

• A subprocess or model expects fixed filenames.
• The same local file contract must hold on cache hits and misses.
• You do not want manual copies to become hidden workflow state.

For lower-level staging helpers, see Materialization.

scenario(...): Multi-Step Workflows¶

Use a scenario when steps belong to one workflow and should be queried together. Each child step gets its own signature, so upstream hits can be reused while downstream misses re-execute.

from pathlib import Path

import pandas as pd

import consist
from consist import ExecutionOptions, Tracker

tracker = Tracker(run_dir="./runs", db_path="./provenance.duckdb")


def clean(raw_path: Path) -> dict[str, Path]:
    df = pd.read_csv(raw_path)
    out = Path("./cleaned.parquet")
    df[df["value"] > 0].to_parquet(out)
    return {"cleaned": out}


def analyze(cleaned_path: Path) -> dict[str, Path]:
    df = pd.read_parquet(cleaned_path)
    out = Path("./analysis.parquet")
    df.groupby("category").size().to_frame("count").to_parquet(out)
    return {"analysis": out}


with tracker.scenario("baseline") as sc:
    cleaned = sc.run(
        name="clean",
        fn=clean,
        inputs={"raw_path": Path("inputs/raw.csv")},
        outputs=["cleaned"],
        execution_options=ExecutionOptions(input_binding="paths"),
    )

    sc.run(
        name="analyze",
        fn=analyze,
        inputs={"cleaned_path": consist.ref(cleaned, key="cleaned")},
        outputs=["analysis"],
        execution_options=ExecutionOptions(input_binding="paths"),
    )

Use scenario(...) when:

• Steps have data dependencies.
• You need per-step cache reuse inside a larger workflow.
• A run should carry shared scenario metadata such as model, year, tags, or facets.
• You need scenario-level output validation.

For decorator defaults, templates, and step metadata, see Decorators and Metadata.

ScenarioContext.map_runs(...): Parallel Sweeps¶

Use map_runs(...) inside a scenario when you have many independent runs with the same callable shape. It is a good fit for parameter sweeps, Monte Carlo replicates, and other wide lineage trees where one parent scenario has many sibling child runs.

from pathlib import Path

import consist
from consist import Tracker

tracker = Tracker(run_dir="./runs", db_path="./provenance.duckdb")

rows = [
    {"case_id": 0, "seed": 100, "scale": 0.8},
    {"case_id": 1, "seed": 101, "scale": 1.0},
    {"case_id": 2, "seed": 102, "scale": 1.2},
]

with tracker.scenario("scale_sweep") as sc:
    batch = sc.map_runs(
        rows=rows,
        fn="my_project.steps:run_case",
        name_template="case_{case_id}",
        model="simulation",
        config_from=lambda row: {"seed": row["seed"], "scale": row["scale"]},
        facet_from=lambda row: {"case_id": row["case_id"], "scale": row["scale"]},
        expected_outputs=["result"],
    )

print(batch.success_count, batch.failure_count)
print(batch.child_run_ids)

The current process backend intentionally requires fn to be an importable "module:function" string. Notebook-local functions, lambdas, and direct callables are not supported for process workers; move the worker into an importable module and pass its string reference.

map_runs(...) returns a BatchResult:

• results: one RunSpecResult per completed or failed spec that returned a result to the parent.
• child_run_ids: persisted child run IDs that can be queried later.
• failed_specs: BatchRunSpec objects for failed rows, suitable for retries.
• retry_rows(): the original config rows for failed specs.

Use cancel_pending_on_failure=True when you want a best-effort attempt to cancel futures that have not started after the first observed failure. Already-running workers may still complete.

ref(...) and refs(...)¶

Use explicit references for step-to-step data flow.

# Select one output from a prior result.
single_input = {"cleaned_path": consist.ref(cleaned, key="cleaned")}

# Reuse output keys as downstream input names.
all_inputs = consist.refs(cleaned)

# Select a subset.
subset = consist.refs(cleaned, "cleaned")

# Rename outputs to match downstream function parameters.
aliased = consist.refs(cleaned, {"cleaned_path": "cleaned"})

Prefer ref(...) for one artifact and refs(...) for reusable mappings. Avoid passing a multi-output RunResult directly as an input because the intended output key is ambiguous.

trace(...): Inline Lifecycle Blocks¶

Use trace(...) when a step does not fit a single callable or when you need to log artifacts directly inside a block.

from pathlib import Path

with tracker.trace(
    name="manual_export",
    model="analysis",
    inputs={"source": Path("inputs/source.parquet")},
) as t:
    output = Path("./export.csv")
    build_export(output)
    t.log_artifact(output, key="export")

Inside a scenario, sc.trace(...) records a child run under the scenario:

with tracker.scenario("baseline") as sc:
    with sc.trace(name="validate", inputs=consist.refs(cleaned, "cleaned")):
        artifact = consist.ref(cleaned, key="cleaned")
        df = consist.load_df(artifact, tracker=tracker)
        assert not df.empty

Important distinction:

• run(...) can skip callable execution on a cache hit.
• trace(...) enters your Python block; use it when explicit lifecycle control matters more than skip-on-hit function execution.

External Tools and Containers¶

For existing tools, keep the wrapper thin:

1. Declare identity-bearing inputs with inputs={...} and depends_on=[...].
2. Use input_binding="paths" so the wrapper receives paths.
3. Use requested input staging only when the tool needs exact local names.
4. Return dict[str, Path] for outputs when possible, or use output_paths.
5. Move container-specific details to the container integration layer.

For container setup, volume behavior, and image identity, see Containers Guide.

Querying Results¶

Start from the tracker when you need to inspect what happened:

run = tracker.get_run(result.run.id)
print(run.status)
print(run.identity_summary)
print(run.meta.get("cache_miss_explanation"))

For indexed metadata and query helpers, see:

• Config Management for config, facets, and queryable metadata.
• Grouped Views for cross-run views over matching artifact schemas.
• CLI Reference for command-line inspection.

Where Deeper Detail Lives¶

• Caching and Hydration: signatures, hit/miss behavior, hydration, and materialization policies.
• Data Storage and Ingestion: ingestion, database fallback, and hybrid views.
• Materialization: hydrate_run_outputs(...), stage_inputs(...), and staging status meanings.
• Run API: accepted run options and generated reference.
• Building a Domain Tracker: when to wrap direct Consist calls in a domain-specific interface.