cursus.core.base.step_interface

StepInterface — the single Pydantic-validated data structure for a .step.yaml.

Combines what was previously ScriptContract + StepSpecification into one validated model. This is the single message passed between dep resolver, builder, and assembler.

It is intentionally a superset of the legacy data classes so it can stand in for all of them during migration:

• StepInterface.contract (ContractSection) is a drop-in for ScriptContract / StepContract: it exposes entry_point, expected_input_paths, expected_output_paths, expected_arguments, required_env_vars, optional_env_vars, framework_requirements and description.

• StepInterface / StepInterface.spec (SpecSection) are a drop-in for StepSpecification: step_type, node_type, dependencies, outputs, get_dependency(), get_output(), get_output_by_name_or_alias(), list_required_dependencies(), list_optional_dependencies(), list_all_output_names(), validate_specification(), validate_contract_alignment() and script_contract.

• Each DependencyDecl / OutputDecl is a drop-in for DependencySpec / OutputSpec: it carries logical_name (auto-populated from the dict key), exposes dependency_type / output_type (aliases of type), and supports matches_name_or_alias().

Validation rules: - Contract inputs and spec dependencies must have matching keys. - Contract outputs and spec outputs must have matching keys. - Paths (when present) must be valid SageMaker paths (processing OR training). - entry_point (when present) must be a .py file.

entry_point and the port path fields are Optional: script-less SageMaker steps (CreateModel / Transform — e.g. xgboost_model, pytorch_model, batch_transform) declare them as null in YAML.

class InputPort(*, path=None, required=True, channels=<factory>)

Bases: BaseModel

One contract input declaration.

path: str | None

required: bool

channels: List[str]

Optional SageMaker training sub-channels this single input fans out into (e.g. [train, val, test]). When set, the TrainingHandler creates one TrainingInput per sub-channel under <path>/<channel>/ instead of a single channel. This is per-step DATA (the channel layout the script expects), so it lives in the .step.yaml — not hardcoded in the handler. Empty/None ⇒ the input maps to a single channel (see TrainingHandler).

classmethod validate_path(v)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OutputPort(*, path=None)

Bases: BaseModel

One contract output declaration.

path: str | None

classmethod validate_path(v)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class EnvVars(*, required=<factory>, optional=<factory>)

Bases: BaseModel

Environment variable declarations.

required: List[str]

optional: Dict[str, str]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class ComputeSpec(*, kind=None, framework_version_field=None, framework_version_default=None, py_version_field=None, sdk_class=None, instance_size_mode='large_or_small', kms_network=False, retrieve_image=False, framework_name=None, lock_training_region=False, locked_region='us-east-1', requires='none')

Bases: BaseModel

Declarative spec for the step’s COMPUTE object — the processor/estimator/model/transformer the builder constructs (FZ 31e1d3k).

Every VALUE is a config field, so the compute object is fully constructable from config + this descriptor — which says WHICH SDK class and WHICH config fields. Lives in the .step.yaml so it is surfaced to users (the steps patterns view) and lets the builder template build the compute generically, replacing the near-identical per-step _create_processor / _create_estimator factories. Empty (kind=None) ⇒ the step keeps its own factory.

kind: str | None

sklearn | xgboost | framework | script (processors) · estimator · model · transformer · None.

framework_version_field: str | None

The config attr holding the framework version (e.g. processing_framework_version).

framework_version_default: str | None

Default framework version when the field is absent on the config — several steps used getattr(config, "processing_framework_version", "<default>") (defaults vary per step, e.g. 1.0-1 / 1.2-1). None ⇒ the field must be present.

py_version_field: str | None

The config attr holding the py version (framework processors / estimators).

sdk_class: str | None

the SDK class NAME to use as the estimator_cls / model class (e.g. PyTorch, SKLearn, XGBoost, PyTorchModel).

Type:

For framework processors / estimators / models

instance_size_mode: str

large_or_small (the use_large_processing_instance ternary) or fixed (a single field).

Type:

How the processing instance type is chosen

kms_network: bool

set the KMS volume key + shared network config + the ECR-from-role image. A genuinely special, declared-once deviation.

Type:

script kind only (EdxUploading)

retrieve_image: bool

explicitly retrieve the training image_uri (PyTorch-for-LightGBM).

Type:

estimator kind only

framework_name: str | None

the framework NAME passed to image_uris.retrieve for the INFERENCE image (xgboost / pytorch). Distinct from sdk_class (the model CLASS, e.g. XGBoostModel): the class instantiates the model, this names the container image to retrieve.

Type:

model kind

lock_training_region: bool

training images/jobs are forced to a fixed region (us-east-1) — an explicit platform constraint, NOT a bug. This is a TOGGLEABLE pattern: a step opts into locking (lock_training_region: true); to run in standard (unlocked) mode it sets lock_training_region: false here (or via config) — no code change. When False, the region comes from config.aws_region (the normal region).

Type:

estimator image-retrieval region locking. SAIS RESTRICTION

locked_region: str

The region used when lock_training_region is True (the SAIS-locked region).

requires: str

DEPENDENCY AXIS — the 3rd-party package this COMPUTE pattern needs at BUILD time (FZ 31e1d3l). none for the sagemaker-only kinds (sklearn/xgboost/framework/estimator/model/transformer); mods_workflow_core ONLY for the script kind with kms_network (the EdxUploading ScriptProcessor, which lazily imports KMS_ENCRYPTION_KEY_PARAM / PROCESSING_JOB_SHARED_NETWORK_CONFIG in builder_base._create_compute). This is a CONSEQUENCE of kms_network — the validator keeps it consistent so it can’t drift, and it is declared in the .step.yaml so the mods-vs-native split is visible (steps patterns).

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

class JobArgDecl(*, flag, source='')

Bases: BaseModel

DECLARATIVE record of one CLI argument the step’s script accepts (FZ 31e1d3h).

Documentation / alignment / introspection only — the TRUE argument list is built at runtime by config.get_job_arguments() (config is the single source). This just makes the script’s argument surface visible in the .step.yaml (the analog of env_vars declaring names).

flag: str

The --flag the script reads (e.g. --job_type, --batch-size).

source: str

The config attribute the value comes from (e.g. job_type). Empty for a bare boolean flag.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class RegistrySection(*, sagemaker_step_type=None, step_assembly=None, requires='none', description='')

Bases: BaseModel

The ‘registry’ section of a .step.yaml — the construction binding + its 3rd-party footprint.

Previously this YAML block was silently dropped (StepInterface had no field for it), so the step_assembly and the create-step dependency had no declaration home. It is now a real section: sagemaker_step_type + step_assembly select the PatternHandler, and requires declares the create_step axis’s BUILD-time 3rd-party dependency.

sagemaker_step_type: str | None

The SageMaker verb that selects the PatternHandler (Processing / Training / CreateModel / Transform / the SAIS verbs). Mirrors the registry’s sagemaker_step_type.

step_assembly: str | None

DEPRECATED — moved to patterns.step_assembly (FZ 31e1d3f1). Kept only as a back-compat read for any not-yet-migrated YAML; _auto_bind_handler + io_view prefer patterns.step_assembly. No .step.yaml in this package declares it here anymore.

requires: str

DEPENDENCY AXIS — the 3rd-party package the CREATE_STEP pattern needs at BUILD time (FZ 31e1d3l). none for the native (sagemaker-only) handlers; secure_ai_sandbox_workflow_python_sdk for the SDKDelegation steps whose builder module imports a SAIS Step class at module level (Registration / CradleDataLoading / DataUploading / RedshiftDataLoading — fatal-on-load if the SDK is absent). Declared here so the mods/SAIS-vs-native split is authored data in the .step.yaml and visible in steps patterns; a conformance gate keeps it equal to the builders’ actual module-level SAIS imports.

description: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

class PatternsSection(*, step_assembly=None, include_job_type_in_path=True, direct_input_keys=<factory>)

Bases: BaseModel

The ‘patterns’ section of a .step.yaml — the per-axis STRATEGY-SELECTION knobs (FZ 31e1d3f1).

This is the BLUEPRINT that guides how the handlers combine/inject behavior per axis, so a step’s implementation is NOT hard-wired in its builder shell. Distinct from contract (script-shaped I/O data), compute (the SDK compute object), registry (discovery + 3rd-party deps), and spec (DAG wiring). _auto_bind_handler reads these into the bound handler’s knobs so editing the YAML steers the build with no Python change.

use_step_args is intentionally NOT a field here: it is DERIVED from step_assembly (the step_args strategy preset sets use_step_args: True, code sets False) — so it can never disagree with the routing verb.

step_assembly: str | None

Processing sub-verb that joins registry.sagemaker_step_type to pick the handler: code (2A, ProcessingStep(code=...)) | step_args (2B, processor.run()) | delegation (SDKDelegation). None ⇒ the handler’s default (code for Processing).

include_job_type_in_path: bool

output_path_token was REMOVED (FZ 31e1d3f1b). The output-destination S3 prefix is DERIVED from the step name — canonical_to_snake(step_type) — not a declarable field: it corresponds to the step name by convention, and the historical deviations were non-standard. Whether config.job_type is a segment of the synthesized output destination.

Type:

NOTE

direct_input_keys: List[str]

Logical input names passed straight through to the processor (not spec×contract joined) — the template-provided direct input allowlist.

as_knobs()

The HANDLER_KNOBS the bound handler reads — only NON-DEFAULT entries, so an unset field falls through to the strategy preset/contract default exactly as the old class-attr knobs did. step_assembly is routing (passed separately to resolve_handler), not a knob.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

class ContractSection(*, entry_point=None, source_dir=False, output_path_token=None, include_job_type_in_path=True, inputs=<factory>, outputs=<factory>, circular_ref_check=False, skip_inputs=<factory>, input_source_overrides=<factory>, sink=False, compute=<factory>, arguments=<factory>, job_arguments=<factory>, env_vars=<factory>, computed_env_paths=<factory>, framework_requirements=<factory>, runtime_requires=<factory>, description='')

Bases: BaseModel

The ‘contract’ section of a .step.yaml — script execution requirements.

Drop-in for the legacy ScriptContract / StepContract: the expected_* / required_env_vars / optional_env_vars accessors flatten the structured ports back to the Dict[str, str] / List[str] shapes consumers expect.

entry_point: str | None

source_dir: bool

Whether this step’s script needs its whole directory uploaded (sibling modules) — i.e. processor.run(code=entry_point, source_dir=<dir>) — vs a self-contained single script processor.run(code=<full_script_path>). This is per-step DATA (a script-packaging fact), so it lives in the .step.yaml rather than being inferred from the processor class. The ProcessingHandler reads it as the split_source_dir switch. Default False (self-contained). NOTE: a True value requires a FrameworkProcessor (ScriptProcessor.run has no source_dir).

output_path_token: str | None

OPT-IN override for the output-destination S3 prefix segment. Default None ⇒ the segment is DERIVED from the step name (canonical_to_snake(step_type)), the convention for ~all steps. When set to a non-empty string it is used VERBATIM as that segment instead of the derived token — needed when an EXTERNAL consumer keys off a fixed S3 folder name that does not match the cursus step name (e.g. PIPER scans <pipeline>/Model_Metric_Generation_Step/ for .metric files). This re-introduces the field removed in FZ 31e1d3f1b, but as an explicit escape hatch (default-off) rather than a routinely-set knob — the derived convention still holds by default.

include_job_type_in_path: bool

Whether config.job_type is a segment of the synthesized output destination. The other _get_outputs axis: some steps put job_type in the path, some don’t. Default True. The ProcessingHandler reads this as the include_job_type_in_path knob.

inputs: Dict[str, InputPort]

outputs: Dict[str, OutputPort]

circular_ref_check: bool

Per-step input-resolution deviations from the standard spec×contract loop (FZ 31e1d3i), read by ProcessingHandler.get_inputs so the step needs no _get_inputs override:

circular_ref_check — run the PipelineVariable circular-reference guard before mapping. skip_inputs — declared dependencies the script loads internally (not mounted as inputs). input_source_overrides {logical_name: config_attr} — take the input SOURCE from a config

attr/method (config is the value source) instead of the resolved dependency value.

skip_inputs: List[str]

input_source_overrides: Dict[str, str]

sink: bool

A SINK step produces no outputs — ProcessingHandler.get_outputs returns [] (FZ 31e1d3i), so a sink step (e.g. an uploader) needs no _get_outputs override.

compute: ComputeSpec

BACK-COMPAT MIRROR of the top-level StepInterface.compute (FZ 31e1d3k). The compute descriptor was promoted to a top-level .step.yaml section (peer of contract/spec) because it describes the BUILDER’s compute object, not the script contract — script-less steps (CreateModel/Transform) have a near-empty contract but a full compute. This field is kept and kept in sync by StepInterface._sync_and_align so existing b.contract.compute read sites still work; authors declare compute: at the top level now.

arguments: Dict[str, str]

job_arguments: List['JobArgDecl']

DECLARATIVE record of the CLI arguments the step’s script accepts (FZ 31e1d3h) — each entry is {flag, source} (the --flag emitted and the config attribute it comes from). This is documentation / alignment / introspection ONLY: the TRUE values are produced at build time by config.get_job_arguments() (config is the single source). Mirrors how env_vars declares names while the config supplies values. Not used to drive _get_job_arguments.

env_vars: EnvVars

computed_env_paths: Dict[str, List[str]]

env vars whose VALUE is an S3 sub-path under the pipeline’s execution prefix (base_output_path), not a config field — e.g. a script that reads/writes an extra staging location. Maps ENV_VAR -> [segment, ...]; the base _get_environment_variables sets ENV_VAR = Join(base_output_path, *segments). This is the declarative form of the formerly-hand-written _get_environment_variables overrides (e.g. BedrockBatchProcessing’s BEDROCK_BATCH_INPUT/OUTPUT_S3_PATH) — the env analog of the output-destination token, so a step needs no Python to compute a runtime S3 env path.

Type:

COMPUTED-S3-ENV pattern (FZ 31e1d3g3 Phase A3)

framework_requirements: Dict[str, str]

runtime_requires: List[str]

DEPENDENCY AXIS (runtime) — 3rd-party packages the step’s SCRIPT imports at CONTAINER runtime (FZ 31e1d3l). This is ORTHOGONAL to build-time deps (compute.requires / registry.requires): these imports live in steps/scripts/<entry_point> and execute inside the SAIS Docker image, NOT during pipeline construction — they never affect offline import of cursus/builders. Kept on a separate descriptor so build-time vs runtime deps are never conflated. E.g. EdxUploading + RedshiftDataLoading scripts import secure_ai_sandbox_python_lib (a runtime, not build, dep).

description: str

classmethod validate_entry_point(v)

property expected_input_paths: Dict[str, str]

property expected_output_paths: Dict[str, str]

property input_channels: Dict[str, List[str]]

Per-input declared training sub-channels (logical_name -> [channel, ...]).

Only inputs that declare a non-empty channels list appear. The TrainingHandler reads this to fan a single input into <path>/<channel>/ sub-channels — the channel layout is per-step DATA in the .step.yaml, not a handler constant.

property expected_arguments: Dict[str, str]

property required_env_vars: List[str]

property optional_env_vars: Dict[str, str]

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class DependencyDecl(*, logical_name='', type=DependencyType.PROCESSING_OUTPUT, required=True, compatible_sources=<factory>, semantic_keywords=<factory>, data_type='S3Uri', description='')

Bases: BaseModel

One spec dependency declaration. Drop-in for the legacy DependencySpec.

logical_name is auto-populated from the dict key by SpecSection’s validator. dependency_type is exposed as an alias of type for the resolver/assembler.

logical_name: str

type: DependencyType

required: bool

compatible_sources: List[str]

semantic_keywords: List[str]

data_type: str

description: str

classmethod coerce_type(v)

property dependency_type: DependencyType

Legacy DependencySpec field name.

matches_name_or_alias(name)

Dependencies have no aliases; matches only the logical name.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class OutputDecl(*, logical_name='', type=DependencyType.PROCESSING_OUTPUT, property_path='', aliases=<factory>, semantic_keywords=<factory>, data_type='S3Uri', description='')

Bases: BaseModel

One spec output declaration. Drop-in for the legacy OutputSpec.

logical_name is auto-populated from the dict key by SpecSection’s validator. output_type is exposed as an alias of type.

logical_name: str

type: DependencyType

property_path: str

aliases: List[str]

semantic_keywords: List[str]

data_type: str

description: str

classmethod coerce_type(v)

property output_type: DependencyType

Legacy OutputSpec field name.

matches_name_or_alias(name)

Check if name matches the logical name or any alias (case-insensitive).

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class SpecSection(*, dependencies=<factory>, outputs=<factory>, step_type='', node_type=NodeType.INTERNAL)

Bases: BaseModel

The ‘spec’ section of a .step.yaml — dependency resolution metadata.

Drop-in for the legacy StepSpecification’s dependency/output surface. The enclosing StepInterface mirrors step_type/node_type here so this object can be registered/consumed standalone where a StepSpecification was expected.

dependencies: Dict[str, DependencyDecl]

outputs: Dict[str, OutputDecl]

step_type: str

node_type: NodeType

get_dependency(logical_name)

get_output(logical_name)

get_output_by_name_or_alias(name)

Get output by logical name or alias (case-insensitive on aliases).

list_all_output_names()

All output logical names plus aliases.

list_required_dependencies()

list_optional_dependencies()

validate_specification()

Consistency check (legacy StepSpecification.validate_specification).

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class VariantDecl(*, spec=<factory>, contract=<factory>)

Bases: BaseModel

A job_type variant block from a .step.yaml (e.g. training / calibration).

Holds the spec/contract overrides that are merged over the base when a builder requests a specific job_type. Stored as raw dicts because they are partial overrides, not standalone sections.

spec: Dict

contract: Dict

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class StepInterface(*, step_type, node_type=NodeType.INTERNAL, registry=<factory>, compute=<factory>, patterns=<factory>, contract, spec=<factory>, variants=<factory>)

Bases: BaseModel

Validated representation of a .step.yaml file.

This is the single message passed among dep resolver, builder, and assembler. Replaces the previous (ScriptContract|StepContract, StepSpecification) tuple.

Build it from a parsed YAML dict via from_yaml(), which applies any requested job_type variant before validation.

step_type: str

node_type: NodeType

registry: RegistrySection

compute: ComputeSpec

Declarative COMPUTE descriptor (FZ 31e1d3k) — a TOP-LEVEL section (peer of contract/spec) because it describes the BUILDER’s compute object (processor/estimator/model/transformer), not the script contract: script-less steps (CreateModel/Transform) carry a near-empty contract but a full compute. _sync_and_align mirrors it onto contract.compute for back-compat. Empty (kind=None) ⇒ the step keeps its own factory.

patterns: PatternsSection

Per-axis STRATEGY-SELECTION knobs (FZ 31e1d3f1) — the blueprint that wires pattern injection (step_assembly / include_job_type_in_path / direct_input_keys), read into the bound handler by _auto_bind_handler so the YAML steers the build, not a builder shell.

contract: ContractSection

spec: SpecSection

variants: Dict[str, VariantDecl]

classmethod coerce_node_type(v)

classmethod from_yaml(data, job_type=None)

Build a StepInterface from a parsed .step.yaml dict, resolving variants.

When job_type names a variant, that variant’s spec/contract overrides are deep-merged over the base sections before validation. The merge is recursive: a variant that lists only a subset of spec.dependencies (or outputs / contract inputs) overrides just those ports’ fields and leaves the rest of the base set intact — it does not replace the whole nested dict. Steps without a matching variant fall back to the base sections unchanged.

A shallow merge here was a latent bug: because variants routinely restate only the ports they tweak, {**base, **variant} at the section level dropped every base port the variant happened to omit (e.g. it dropped hyperparameters_s3_uri from RiskTableMapping’s variants, which then violated the contract↔spec alignment invariant and raised at construction).

property script_contract: ContractSection

Legacy StepSpecification.script_contract accessor.

property entry_point: str | None

property expected_input_paths: Dict[str, str]

property expected_output_paths: Dict[str, str]

property expected_arguments: Dict[str, str]

property required_env_vars: List[str]

property optional_env_vars: Dict[str, str]

property framework_requirements: Dict[str, str]

property description: str

property dependencies: Dict[str, DependencyDecl]

property outputs: Dict[str, OutputDecl]

get_dependency(logical_name)

get_output(logical_name)

get_output_by_name_or_alias(name)

list_all_output_names()

list_required_dependencies()

list_optional_dependencies()

validate_specification()

validate_contract_alignment()

Validate that the contract aligns with the spec.

Mirrors legacy StepSpecification.validate_contract_alignment: every contract input must have a matching spec dependency, and every contract output a matching spec output (extra spec deps/outputs and output aliases allowed). Returns a ValidationResult (is_valid / errors).

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].