Skip to content

Step Contracts

Step contracts expose resolved @define_step metadata without executing the step. Use them when an external orchestrator, schema builder, or validation layer needs the same declarative metadata that Tracker.run(...) and ScenarioContext.run(...) would use at execution time.

API reference

consist.core.step_contracts.StepContract dataclass

Resolved metadata contract for a @define_step function.

This is the planning/introspection counterpart to the run invocation metadata used by Tracker.run and ScenarioContext.run. It intentionally carries the declarative fields that orchestrators need before execution without owning execution, output registration, or domain-specific coupler policy.

Attributes:

Name Type Description
name str

Resolved step run name after explicit overrides, decorator name_template, and default name templates are applied.
model str

Resolved model name that execution would persist on the run record. When no explicit model is provided, this falls back to name.
description str or None

Resolved human-readable step description.
config dict[str, Any] or None

Resolved configuration payload that would contribute to run identity.
adapter Any or None

Resolved config adapter, when decorator metadata or an explicit override provides one.
identity_inputs IdentityInputs

Resolved hash-only identity inputs. These are paths or labeled paths that affect run identity without being logged as artifacts.
tags list[str] or None

Resolved run tags.
facet FacetLike or None

Resolved queryable facet payload.
facet_from list[str] or None

Config keys that execution would project into the facet payload.
facet_schema_version str, int, or None

Optional schema version for the resolved facet payload.
facet_index bool or None

Whether scalar facet values should be indexed for query.
inputs mapping or iterable or None

Resolved direct input references before execution-time materialization or coupler lookup.
input_keys iterable[str], str, or None

Required coupler input keys resolved from decorator or call-site metadata.
optional_input_keys iterable[str], str, or None

Optional coupler input keys resolved from decorator or call-site metadata.
outputs list[str] or None

Resolved Python return-value output keys.
output_paths mapping[str, ArtifactRef] or None

Resolved key-to-path mapping for file-writing steps.
output_sets mapping[str, OutputSet] or None

Resolved logical multi-file outputs.
cache_mode str or None

Resolved cache mode, when explicitly supplied by metadata or options.
cache_hydration str or None

Resolved cache-hit output hydration policy.
cache_version int or None

Resolved per-step cache version.
validate_cached_outputs str or None

Resolved cached-output validation policy.
input_binding {'loaded', 'paths', 'none'} or None

Resolved callable input binding mode.
load_inputs bool or None

Resolved legacy input-loading preference.

adapter instance-attribute

cache_hydration instance-attribute

cache_mode instance-attribute

cache_version instance-attribute

config instance-attribute

description instance-attribute

facet instance-attribute

facet_from instance-attribute

facet_index instance-attribute

facet_schema_version instance-attribute

identity_inputs instance-attribute

input_binding instance-attribute

input_keys instance-attribute

inputs instance-attribute

load_inputs instance-attribute

model instance-attribute

name instance-attribute

optional_input_keys instance-attribute

output_paths instance-attribute

output_sets instance-attribute

outputs instance-attribute

tags instance-attribute

validate_cached_outputs instance-attribute

__init__(name, model, description, config, adapter, identity_inputs, tags, facet, facet_from, facet_schema_version, facet_index, inputs, input_keys, optional_input_keys, outputs, output_paths, output_sets, cache_mode, cache_hydration, cache_version, validate_cached_outputs, input_binding, load_inputs)

consist.core.step_contracts.resolve_step_contract(step, *, name=None, model=None, description=None, config=None, adapter=None, identity_inputs=None, inputs=None, input_keys=None, optional_input_keys=None, tags=None, facet=None, facet_from=None, facet_schema_version=None, facet_index=None, year=None, iteration=None, phase=None, stage=None, outputs=None, output_paths=None, output_sets=None, cache_options=None, execution_options=None, default_name_template=None, allow_template=True, apply_step_defaults=True, consist_settings=None, consist_workspace=None, consist_state=None, runtime_kwargs=None, missing_name_error='resolve_step_contract requires a step name.')

Resolve a decorated step's declarative metadata without executing it.

Explicit arguments use the same precedence as run execution: call-site values override decorator metadata, while callable decorator metadata is resolved with a StepContext populated from the supplied workflow context.

Parameters:

Name Type Description Default
step Callable[..., Any]

Step function to inspect. If the function was decorated with @define_step, its metadata is applied using the same precedence as Tracker.run and ScenarioContext.run.

 required
name str

Explicit step name override.

 None
model str

Explicit model name override.

 None
description str

Explicit description override.

 None
config dict[str, Any]

Explicit configuration payload override.

 None
adapter Any

Explicit config adapter override.

 None
identity_inputs IdentityInputs

Explicit hash-only identity inputs.

 None
inputs mapping or iterable

Direct input references to include in the resolved contract.

 None
input_keys iterable[str] or str

Required coupler input keys.

 None
optional_input_keys iterable[str] or str

Optional coupler input keys.

 None
tags list[str]

Explicit run tags.

 None
facet FacetLike

Explicit queryable facet payload.

 None
facet_from list[str]

Config keys to project into facets during execution.

 None
facet_schema_version str or int

Facet schema version.

 None
facet_index bool

Whether scalar facet values should be indexed.

 None
year int

Workflow year made available to callable metadata and name templates.

 None
iteration int

Workflow iteration made available to callable metadata and name templates.

 None
phase str

Workflow phase made available to callable metadata and name templates.

 None
stage str

Workflow stage made available to callable metadata and name templates.

 None
outputs list[str]

Explicit return-value output key override.

 None
output_paths mapping[str, ArtifactRef]

Explicit file output path mapping.

 None
output_sets mapping[str, OutputSet]

Explicit logical multi-file output mapping.

 None
cache_options CacheOptions

Cache policy options whose fields are resolved into the contract.

 None
execution_options ExecutionOptions

Execution options whose metadata-relevant fields are resolved into the contract. Runtime kwargs may be supplied here or through runtime_kwargs, but not both.

 None
default_name_template str

Name template used when the step decorator does not provide one.

 None
allow_template bool

Whether name templates are applied.

 True
apply_step_defaults bool

Whether @define_step metadata defaults are applied.

 True
consist_settings ConsistSettings

Consist settings object exposed to callable metadata as ctx.consist_settings.

 None
consist_workspace Path

Consist run workspace exposed to callable metadata as ctx.consist_workspace.

 None
consist_state ConsistRecord

Active Consist state exposed to callable metadata as ctx.consist_state.

 None
runtime_kwargs mapping[str, Any]

Workflow runtime objects exposed to callable metadata through ctx.runtime_kwargs and the ctx.runtime_* convenience fields.

 None
missing_name_error str

Error message used when neither an explicit name, function name, nor template can resolve a name.

 "resolve_step_contract requires a step name."

Returns:

Type Description
StepContract

Fully resolved declarative step metadata.

Raises:

Type Description
ValueError

If runtime kwargs are provided both directly and through execution_options.runtime_kwargs.
ValueError

If the step name cannot be resolved.

Examples:

>>> import consist
>>> @consist.define_step(
...     outputs=lambda ctx: [f"results_{ctx.year}"],
...     name_template="{func_name}__y{year}",
... )
... def simulate():
...     pass
>>> contract = consist.resolve_step_contract(simulate, year=2030)
>>> contract.name
'simulate__y2030'
>>> contract.outputs
['results_2030']

consist.core.step_contracts.collect_step_contracts(steps, **context)

Resolve a sequence of decorated steps with shared workflow context.

Parameters:

Name Type Description Default
steps iterable of Callable[..., Any]

Step functions to inspect.

 required
**context Any

Shared keyword arguments forwarded to resolve_step_contract for each step, such as year, iteration, runtime_kwargs, cache_options, or default_name_template.

 {}

Returns:

Type Description
list[StepContract]

Resolved contracts in the same order as steps.

Examples:

>>> import consist
>>> @consist.define_step(outputs=lambda ctx: [f"a_{ctx.year}"])
... def first():
...     pass
>>> contracts = consist.collect_step_contracts([first], year=2040)
>>> contracts[0].outputs
['a_2040']