cursus.core

Cursus Core module.

This module provides the core functionality for Cursus, including: - Pipeline assembling and template management - DAG compilation and configuration resolution - Configuration field management and three-tier architecture - Dependency resolution and specification management - Base classes for configurations, contracts, specifications, and builders

class DependencyType(*values)

Bases: Enum

Types of dependencies in the pipeline.

MODEL_ARTIFACTS = 'model_artifacts'

PROCESSING_OUTPUT = 'processing_output'

TRAINING_DATA = 'training_data'

HYPERPARAMETERS = 'hyperparameters'

PAYLOAD_SAMPLES = 'payload_samples'

CUSTOM_PROPERTY = 'custom_property'

class NodeType(*values)

Bases: Enum

Types of nodes in the pipeline based on their dependency/output characteristics.

SOURCE = 'source'

INTERNAL = 'internal'

SINK = 'sink'

SINGULAR = 'singular'

class ValidationResult(*, is_valid, errors=<factory>, warnings=<factory>)

Bases: BaseModel

Result of script contract validation

add_error(error)

Add an error to the result and mark as invalid

add_warning(warning)

Add a warning to the result

classmethod combine(results)

Combine multiple validation results

classmethod error(errors)

Create a failed validation result

model_config: ClassVar[ConfigDict] = {}

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

classmethod success(message='Validation passed')

Create a successful validation result

is_valid: bool

errors: List[str]

warnings: List[str]

class ScriptAnalyzer(script_path)

Bases: object

Analyzes Python scripts to extract I/O patterns and environment variable usage

property ast_tree: Any

Lazy load and parse the script AST

get_argument_usage()

Extract command-line arguments used by the script

get_env_var_usage()

Extract environment variables accessed by the script

get_input_paths()

Extract input paths used by the script

get_output_paths()

Extract output paths used by the script

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.

classmethod coerce_node_type(v)

property dependencies: Dict[str, DependencyDecl]

property description: str

property entry_point: str | None

property expected_arguments: Dict[str, str]

property expected_input_paths: Dict[str, str]

property expected_output_paths: Dict[str, str]

property framework_requirements: Dict[str, str]

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

get_dependency(logical_name)

get_output(logical_name)

get_output_by_name_or_alias(name)

list_all_output_names()

list_optional_dependencies()

list_required_dependencies()

model_config: ClassVar[ConfigDict] = {}

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

property optional_env_vars: Dict[str, str]

property outputs: Dict[str, OutputDecl]

property required_env_vars: List[str]

property script_contract: ContractSection

Legacy StepSpecification.script_contract accessor.

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

validate_specification()

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]

class ModelHyperparameters(*, full_field_list, cat_field_list, tab_field_list, id_name, label_name, multiclass_categories, categorical_features_to_encode=<factory>, model_class='base_model', device=-1, header=0, lr=3e-05, batch_size=2, eval_batch_size_multiplier=2.0, max_epochs=3, metric_choices=['f1_score', 'auroc'], optimizer='SGD', class_weights=None, **extra_data)

Bases: BaseModel

Base model hyperparameters for training tasks.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

categorize_fields()

Categorize all fields into three tiers: 1. Tier 1: Essential User Inputs - fields with no defaults (required) 2. Tier 2: System Inputs - fields with defaults (optional) 3. Tier 3: Derived Fields - properties that access private attributes

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names

Return type:

Dict[str, List[str]]

classmethod from_base_hyperparam(base_hyperparam, **kwargs)

Create a new hyperparameter instance from a base hyperparameter. This is a virtual method that all derived classes can use to inherit from a parent config.

Parameters:

• base_hyperparam (ModelHyperparameters) – Parent ModelHyperparameters instance

• **kwargs (Any) – Additional arguments specific to the derived class

Returns:

A new instance of the derived class initialized with parent fields and additional kwargs

Return type:

ModelHyperparameters

get_config()

Get the complete configuration dictionary.

get_public_init_fields()

Get a dictionary of public fields suitable for initializing a child hyperparameter. Only includes fields that should be passed to child class constructors. Both essential user inputs and system inputs with defaults or user-overridden values are included to ensure all user customizations are properly propagated.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

property input_tab_dim: int

Get input tabular dimension derived from tab_field_list.

property is_binary: bool

Determine if this is a binary classification task based on num_classes.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow', 'protected_namespaces': (), 'validate_assignment': True}

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.

property num_classes: int

Get number of classes derived from multiclass_categories.

print_hyperparam()

Print complete hyperparameter information organized by tiers. This method automatically categorizes fields by examining their characteristics.

serialize_config()

Serialize configuration for SageMaker.

validate_dimensions()

Validate model dimensions and configurations

full_field_list: List[str]

cat_field_list: List[str]

tab_field_list: List[str]

id_name: str

label_name: str

multiclass_categories: List[str | int]

categorical_features_to_encode: List[str]

model_class: str

device: int

header: int

lr: float

batch_size: int

eval_batch_size_multiplier: float

max_epochs: int

metric_choices: List[str]

optimizer: str

class_weights: List[float] | None

class BasePipelineConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.0', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=False, max_runtime_seconds=172800, project_root_folder, **extra_data)

Bases: BaseModel, ABC

Base configuration with shared pipeline attributes and self-contained derivation logic.

property aws_region: str

Get AWS region based on region code.

categorize_fields()

Categorize all fields into three tiers: 1. Tier 1: Essential User Inputs - public fields with no defaults (required) 2. Tier 2: System Inputs - public fields with defaults (optional) 3. Tier 3: Derived Fields - properties that access private attributes

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names

Return type:

Dict[str, List[str]]

property effective_source_dir: str | None

Get effective source directory with hybrid resolution.

This base implementation works with just source_dir (which can be None). Processing configs override this to handle both processing_source_dir and source_dir.

Resolution Priority: 1. Hybrid resolution of source_dir 2. Legacy value (source_dir) 3. None if source_dir is not provided

classmethod from_base_config(base_config, **kwargs)

Create a new configuration instance from a base configuration. This is a virtual method that all derived classes can use to inherit from a parent config.

Parameters:

• base_config (BasePipelineConfig) – Parent BasePipelineConfig instance

• **kwargs (Any) – Additional arguments specific to the derived class

Returns:

A new instance of the derived class initialized with parent fields and additional kwargs

Return type:

BasePipelineConfig

classmethod get_config_class_name(step_name)

Get the configuration class name from a step name using existing registry functions.

get_environment_variables(declared_env_vars=None)

Resolve the env-var VALUES for the names the step interface DECLARES (FZ 31e1d3g).

The interface (.step.yaml env_vars) declares WHICH env vars a step uses; this config supplies the VALUES. The single-source rule: the interface drives the key set, config resolves each one — by _env_overrides() first (computed/aliased), else convention (NAME -> self.name, type-formatted). Names that resolve to nothing are omitted, so a declared-but-absent optional falls back to its interface default in the builder.

Subclasses with a bespoke collector may override this entirely (and ignore declared_env_vars) to return their full env dict — that path is preserved for back-compat.

get_job_arguments()

Build the script’s CLI arguments — config is the single source (FZ 31e1d3h).

Base default: no CLI arguments (None). The builder’s _get_job_arguments calls this; a step-config that passes arguments to its script OVERRIDES this method. The common --job_type case has a ready helper, _job_type_arg() — a config opts in with return self._job_type_arg(). None (not []) is the “no args” contract the SDK ProcessingStep / processor.run expect.

get_public_init_fields()

Get a dictionary of public fields suitable for initializing a child config. Only includes fields that should be passed to child class constructors. Both essential user inputs and system inputs with defaults or user-overridden values are included to ensure all user customizations are properly propagated to derived classes.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_contract()

Get script contract for this configuration using optimized step catalog discovery.

This optimized implementation uses the step catalog system for efficient contract discovery with O(1) lookups and workspace awareness, falling back to legacy methods for backward compatibility.

Returns:

Script contract instance or None if not available

Return type:

Any | None

get_script_path(default_path=None)

Get script path for this configuration.

This method provides a default implementation that returns None, since not all step types require scripts (e.g., Model creation steps don’t need scripts).

Derived classes that need script paths should override this method with their specific requirements: - Processing steps: Combine entry_point with source_dir - Training steps: Use contract entry_point or combine with source_dir - Model steps: May not need scripts at all

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path, default_path, or None if not applicable for this step type

Return type:

str | None

classmethod get_step_name(config_class_name)

Get the step name for a configuration class using existing registry functions.

initialize_derived_fields()

Initialize all derived fields once after validation.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow', 'protected_namespaces': (), 'validate_assignment': True}

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property pipeline_description: str

Get pipeline description derived from service_name, model_class, and region.

property pipeline_name: str

Get pipeline name derived from author, service_name, model_class, and region.

property pipeline_s3_loc: str

Get S3 location for pipeline artifacts.

print_config()

Print complete configuration information organized by tiers. This method automatically categorizes fields by examining their characteristics: - Tier 1: Essential User Inputs (public fields without defaults) - Tier 2: System Inputs (public fields with defaults) - Tier 3: Derived Fields (properties that provide access to private fields)

resolve_hybrid_path(relative_path)

Resolve a path using the hybrid path resolution system.

This method uses the hybrid path resolution system to find files across different deployment scenarios (Lambda/MODS bundled, development monorepo, pip-installed separated).

Parameters:

relative_path (str) – Relative path from project root to target

Returns:

Resolved absolute path if found, None otherwise

Return type:

str | None

property resolved_source_dir: str | None

Get resolved source directory using hybrid resolution.

Returns None if source_dir is not provided, since it’s optional in base class. Processing, training, and model step configs should ensure source_dir is provided.

property script_contract: Any | None

Property accessor for script contract.

Returns:

Script contract instance or None if not available

property step_catalog: Any | None

Lazy-loaded step catalog instance for optimized component discovery.

Returns:

StepCatalog instance or None if initialization fails

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

framework_version: str

py_version: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class StepBuilderBase(config, spec=None, sagemaker_session=None, role=None, registry_manager=None, dependency_resolver=None)

Bases: ABC

Base class for all step builders

## Safe Logging Methods

To handle Pipeline variables safely in logs, use these methods:

```python # Instead of: logger.info(f”Using input path: {input_path}”) # May raise TypeError for Pipeline variables

# Use: self.log_info(“Using input path: %s”, input_path) # Handles Pipeline variables safely ```

Standard Pattern for input_names and output_names:

1. In config classes: `python output_names = {"logical_name": "DescriptiveValue"}  # VALUE used as key in outputs dict input_names = {"logical_name": "ScriptInputName"}    # KEY used as key in inputs dict `

2. In pipeline code: ```python # Get output using VALUE from output_names output_value = step_a.config.output_names[“logical_name”] output_uri = step_a.properties.ProcessingOutputConfig.Outputs[output_value].S3Output.S3Uri

   # Set input using KEY from input_names inputs = {“logical_name”: output_uri} ```

3. In step builders: ```python # For outputs - validate using VALUES value = self.config.output_names[“logical_name”] if value not in outputs:

   raise ValueError(f”Must supply an S3 URI for ‘{value}’”)

   # For inputs - validate using KEYS for logical_name in self.config.input_names.keys():

   if logical_name not in inputs:

   raise ValueError(f”Must supply an S3 URI for ‘{logical_name}’”)

   ```

Developers should follow this standard pattern when creating new step builders. The base class provides helper methods to enforce and simplify this pattern:

• _validate_inputs(): Validates inputs using KEYS from input_names

• _validate_outputs(): Validates outputs using VALUES from output_names

• _get_script_input_name(): Maps logical name to script input name

• _get_output_destination_name(): Maps logical name to output destination name

• _create_standard_processing_input(): Creates standardized ProcessingInput

• _create_standard_processing_output(): Creates standardized ProcessingOutput

Property Path Registry:

To bridge the gap between definition-time and runtime, step builders can register property paths that define how to access their outputs at runtime. This solves the issue where outputs are defined statically but only accessible via specific runtime paths.

• register_property_path(): Registers a property path for a logical output name

• get_property_paths(): Gets all registered property paths for this step

COMMON_PROPERTIES = {'dependencies': 'Optional list of dependent steps', 'enable_caching': 'Whether to enable caching for this step (default: True)'}

MODEL_OUTPUT_PROPERTIES = {'model': 'SageMaker model object', 'model_artifacts_path': 'S3 path to model artifacts'}

REGION_MAPPING: Dict[str, str] = {'EU': 'eu-west-1', 'FE': 'us-west-2', 'NA': 'us-east-1'}

STEP_NAME: str | None = None

The canonical registry step name for THIS builder (singular), e.g. “XGBoostTraining”. This is the authoritative identity slot _get_step_name reads (FZ 31e1d3g3 Phase C1): every routed shell + TemplateStepBuilder sets it, the materializer stamps it on synthesized fileless builders, and once the per-step shell classes are deleted it is the ONLY reliable canonical key (the class name collapses to TemplateStepBuilder). Declared here on the root so the base method that reads it owns the slot, and so it is not confused with STEP_NAMES (PLURAL — the whole registry dict, below). None on a hand-written builder, which falls back to the legacy <Name>StepBuilder class-name convention.

property STEP_NAMES: Dict[str, Any]

Lazy load step names with workspace context awareness.

This property now supports workspace-aware step name resolution by: 1. Extracting workspace context from config or environment 2. Using hybrid registry manager for workspace-specific step names 3. Falling back to traditional registry if hybrid is unavailable 4. Maintaining backward compatibility with existing code

Returns:

Step names mapping for the current workspace context

Return type:

Dict[str, str]

TRAINING_OUTPUT_PROPERTIES = {'model_data': 'S3 path to the model artifacts', 'model_data_url': 'S3 URL to the model artifacts', 'training_job_name': 'Name of the training job'}

abstractmethod create_step(**kwargs)

Create pipeline step.

This method should be implemented by all step builders to create a SageMaker pipeline step. It accepts a dictionary of keyword arguments that can be used to configure the step.

Common parameters that all step builders should handle: - dependencies: Optional list of steps that this step depends on - enable_caching: Whether to enable caching for this step (default: True)

Step-specific parameters should be extracted from kwargs as needed.

Parameters:

**kwargs (Any) – Keyword arguments for configuring the step

Returns:

SageMaker pipeline step

Return type:

Step

extract_inputs_from_dependencies(dependency_steps)

Extract inputs from dependency steps using the UnifiedDependencyResolver.

Parameters:

dependency_steps (List[Step]) – List of dependency steps

Returns:

Dictionary of inputs extracted from dependency steps

Raises:

ValueError – If dependency resolver is not available or specification is not provided

Return type:

Dict[str, Any]

get_all_property_paths()

Get all property paths defined in the specification.

Returns:

Mapping from logical output names to runtime property paths

Return type:

dict

get_optional_dependencies()

Get list of optional dependency logical names from specification.

This method provides direct access to the optional dependencies defined in the step specification.

Returns:

List of logical names for optional dependencies

Raises:

ValueError – If specification is not provided

Return type:

List[str]

get_outputs()

Get output specifications directly from the step specification.

This method provides direct access to the outputs defined in the step specification, returning the complete OutputSpec objects.

Returns:

Dictionary mapping output names to their OutputSpec objects

Raises:

ValueError – If specification is not provided

Return type:

Dict[str, Any]

get_property_path(logical_name, format_args=None)

Get property path for an output using the specification.

This method retrieves the property path for an output from the specification. It also supports template formatting if format_args are provided.

Parameters:

• logical_name (str) – Logical name of the output

• format_args (Dict[str, Any] | None) – Optional dictionary of format arguments for template paths (e.g., {‘output_descriptor’: ‘data’} for paths with placeholders)

Returns:

Property path from specification, formatted with args if provided, or None if not found

Return type:

str | None

get_required_dependencies()

Get list of required dependency logical names from specification.

This method provides direct access to the required dependencies defined in the step specification.

Returns:

List of logical names for required dependencies

Raises:

ValueError – If specification is not provided

Return type:

List[str]

log_debug(message, *args, **kwargs)

Debug version of safe logging

log_error(message, *args, **kwargs)

Error version of safe logging

log_info(message, *args, **kwargs)

Safely log info messages, handling Pipeline variables.

Parameters:

• message (str) – The log message

• *args (Any) – Values to format into the message

• **kwargs (Any) –

  Values to format into the message

log_warning(message, *args, **kwargs)

Warning version of safe logging

set_execution_prefix(execution_prefix=None)

Set the execution prefix for dynamic output path resolution.

This method is called by PipelineAssembler to provide the execution prefix that step builders use for dynamic output path generation.

Based on analysis of regional_xgboost.py, only PIPELINE_EXECUTION_TEMP_DIR is used by step builders for output paths. Other pipeline parameters (KMS_ENCRYPTION_KEY_PARAM, VPC_SUBNET, SECURITY_GROUP_ID) are used at the pipeline level, not in step builders.

Parameters:

execution_prefix (str | Any | None) – The execution prefix that can be either: - ParameterString: PIPELINE_EXECUTION_TEMP_DIR from pipeline parameters - str: config.pipeline_s3_loc as fallback - None: No parameter found, will fall back to config.pipeline_s3_loc

validate_configuration()

Validate builder-context configuration requirements (optional hook).

No-op by default. The Pydantic config class is the authority for config validation — required fields, @field_validator / @model_validator constraints, and defaults are all enforced at config construction, BEFORE the builder runs. A config that constructs is valid by definition, so most builders need no override here (FZ 31e1d3e). Override ONLY to assert an invariant the config genuinely cannot express — one involving builder context (self.role / self.session / self.spec / resolved dependencies) or a cross-field rule not yet on the config model.

class PipelineAssembler(dag, config_map, step_catalog=None, sagemaker_session=None, role=None, pipeline_parameters=None, registry_manager=None, dependency_resolver=None)

Bases: object

Assembles pipeline steps using a DAG and step builders with specification-based dependency resolution.

This class implements a component-based approach to building SageMaker Pipelines, leveraging the specification-based dependency resolution system to simplify the code and improve maintainability.

The assembler follows these steps to build a pipeline: 1. Initialize step builders for all steps in the DAG 2. Determine the build order using topological sort 3. Propagate messages between steps using the dependency resolver 4. Instantiate steps in topological order, delegating input/output handling to builders 5. Create the pipeline with the instantiated steps

This approach allows for a flexible and modular pipeline definition, where each step is responsible for its own configuration and input/output handling.

analyze_pipeline_structure()

Analyze and print the complete pipeline structure including: 1. Dependency graph between steps 2. Input assignments with logical names, property paths, and destination paths

This method uses internal assembler data (step_messages, step_builders, step_instances) to provide comprehensive insights into how the pipeline is structured.

This should be called after generate_pipeline() to ensure all steps are instantiated.

classmethod create_with_components(dag, config_map, step_catalog=None, context_name=None, **kwargs)

Create pipeline assembler with managed components.

This factory method creates a pipeline assembler with properly configured dependency components from the factory module.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance defining the pipeline structure

• config_map (Dict[str, BasePipelineConfig]) – Mapping from step name to config instance

• step_catalog (StepCatalog | None) – StepCatalog for config-to-builder resolution

• context_name (str | None) – Optional context name for registry

• **kwargs (Any) – Additional arguments to pass to the constructor

Returns:

Configured PipelineAssembler instance

Return type:

PipelineAssembler

generate_pipeline(pipeline_name)

Build and return a SageMaker Pipeline object.

This method builds the pipeline by: 1. Propagating messages between steps using specification-based matching 2. Instantiating steps in topological order 3. Creating the pipeline with the instantiated steps

Parameters:

pipeline_name (str) – Name of the pipeline

Returns:

SageMaker Pipeline object

Return type:

Pipeline

generate_single_node_pipeline(target_node, manual_inputs, pipeline_name)

Generate pipeline with single node and manual inputs.

This method bypasses normal message propagation and dependency resolution, directly instantiating the target node with provided manual inputs.

Backward Compatibility: This method is completely isolated and does not affect existing pipeline generation flow. The normal generate_pipeline() method remains unchanged.

Parameters:

• target_node (str) – Name of node to execute in isolation

• manual_inputs (Dict[str, str]) – Manual input paths (logical_name -> s3_uri)

• pipeline_name (str) – Name for generated pipeline

Returns:

Single-node Pipeline

Raises:

ValueError – If target node not found in step builders

Return type:

Pipeline

Example

>>> manual_inputs = {
...     "input_path": "s3://bucket/run-123/preprocess/data/"
... }
>>> pipeline = assembler.generate_single_node_pipeline(
...     target_node="train",
...     manual_inputs=manual_inputs,
...     pipeline_name="train-debug-001"
... )

class PipelineTemplateBase(config_path, sagemaker_session=None, role=None, registry_manager=None, dependency_resolver=None, pipeline_parameters=None, step_catalog=None)

Bases: ABC

Base class for all pipeline templates.

This class provides a consistent structure and common functionality for all pipeline templates, enforcing best practices and ensuring proper component lifecycle management.

The template follows these steps to build a pipeline: 1. Load configurations from file 2. Initialize component dependencies (registry_manager, dependency_resolver) 3. Create the DAG, config_map, and step_builder_map 4. Use PipelineAssembler to assemble the pipeline

This provides a standardized approach for creating pipeline templates, reducing code duplication and enforcing best practices.

CONFIG_CLASSES: Dict[str, Type[BasePipelineConfig]] = {}

analyze_pipeline_structure()

Analyze and print the complete pipeline structure.

Delegates to the PipelineAssembler’s analyze_pipeline_structure method. Must be called after generate_pipeline().

Raises:

AttributeError – If called before generate_pipeline()

classmethod build_in_thread(config_path, **kwargs)

Build pipeline using thread-local component instances.

This method creates a template with thread-local component instances, ensuring thread safety in multi-threaded environments.

Parameters:

• config_path (str) – Path to configuration file

• **kwargs (Any) – Additional arguments to pass to constructor

Returns:

Generated pipeline

Return type:

Pipeline

classmethod build_with_context(config_path, **kwargs)

Build pipeline with scoped dependency resolution context.

This method creates a template with a dependency resolution context that ensures proper cleanup of resources after pipeline generation.

Parameters:

• config_path (str) – Path to configuration file

• **kwargs (Any) – Additional arguments to pass to constructor

Returns:

Generated pipeline

Return type:

Pipeline

classmethod create_with_components(config_path, context_name=None, **kwargs)

Create template with managed dependency components.

This factory method creates a template with properly configured dependency resolution components from the factory module.

Parameters:

• config_path (str) – Path to configuration file

• context_name (str | None) – Optional context name for registry isolation

• **kwargs (Any) – Additional arguments to pass to constructor

Returns:

Template instance with managed components

Return type:

PipelineTemplateBase

generate_pipeline()

Generate the SageMaker Pipeline.

This method coordinates the pipeline generation process: 1. Create the DAG and config_map (the assembler self-discovers builders) 2. Create the PipelineAssembler 3. Generate the pipeline 4. Store pipeline metadata

Returns:

SageMaker Pipeline

Return type:

Pipeline

set_pipeline_parameters(parameters=None)

Set pipeline parameters for this template.

This method allows DAGCompiler to inject custom parameters that will be used instead of the default parameters defined in subclasses.

Parameters:

parameters (List[ParameterString] | None) – List of pipeline parameters to use

compile_dag_to_pipeline(dag=None, dag_path=None, config_path=None, sagemaker_session=None, role=None, pipeline_name=None, **kwargs)

Compile a PipelineDAG into a complete SageMaker Pipeline.

This is the main entry point for users who want a simple, one-call compilation from DAG to pipeline.

Parameters:

• dag (PipelineDAG | None) – PipelineDAG instance defining the pipeline structure (optional if dag_path provided)

• dag_path (str | None) – Path to serialized DAG JSON file (optional if dag provided)

• config_path (str) – Path to configuration file containing step configs

• sagemaker_session (PipelineSession | None) – SageMaker session for pipeline execution

• role (str | None) – IAM role for pipeline execution

• pipeline_name (str | None) – Optional pipeline name override

• **kwargs (Any) – Additional arguments passed to template constructor

Returns:

Generated SageMaker Pipeline ready for execution

Raises:

• ValueError – If neither dag nor dag_path provided, or if DAG is invalid

• FileNotFoundError – If dag_path or config_path files not found

• ConfigurationError – If configuration validation fails

• RegistryError – If step builders not found for config types

Return type:

Pipeline

Example

>>> # Option 1: Using PipelineDAG instance
>>> dag = PipelineDAG()
>>> dag.add_node("data_load")
>>> dag.add_node("preprocess")
>>> dag.add_edge("data_load", "preprocess")
>>>
>>> pipeline = compile_dag_to_pipeline(
...     dag=dag,
...     config_path="configs/my_pipeline.json",
...     sagemaker_session=session,
...     role="arn:aws:iam::123456789012:role/SageMakerRole"
... )
>>> pipeline.upsert()
>>>
>>> # Option 2: Loading DAG from file
>>> pipeline = compile_dag_to_pipeline(
...     dag_path="saved_dags/my_pipeline.json",
...     config_path="configs/my_pipeline.json",
...     sagemaker_session=session,
...     role="arn:aws:iam::123456789012:role/SageMakerRole"
... )

class PipelineDAGCompiler(config_path, sagemaker_session=None, role=None, config_resolver=None, step_catalog=None, pipeline_parameters=None, project_root=None, anchor_file=None, workspace_dirs=None, **kwargs)

Bases: object

Advanced API for DAG-to-template compilation with additional control.

This class provides more control over the compilation process, including validation, debugging, and customization options.

analyze_pipeline_structure()

Analyze and print the complete pipeline structure.

Delegates to the template’s analyze_pipeline_structure method. Must be called after compile() or compile_with_report().

Raises:

AttributeError – If called before compilation

compile(dag, pipeline_name=None, **kwargs)

Compile DAG to pipeline with full control.

Parameters:

• dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file

• pipeline_name (str | None) – Optional pipeline name override

• **kwargs (Any) – Additional arguments for template

Returns:

Generated SageMaker Pipeline

Raises:

PipelineAPIError – If compilation fails

Return type:

Pipeline

compile_with_report(dag, pipeline_name=None, **kwargs)

Compile DAG to pipeline and return detailed compilation report.

Parameters:

• dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file

• pipeline_name (str | None) – Optional pipeline name override

• **kwargs (Any) – Additional arguments for template

Returns:

Tuple of (Pipeline, ConversionReport)

Return type:

Tuple[Pipeline, ConversionReport]

create_template(dag, **kwargs)

Create a pipeline template from the DAG without generating the pipeline.

This allows inspecting or modifying the template before pipeline generation.

Parameters:

• dag (PipelineDAG) – PipelineDAG instance to create a template for

• **kwargs (Any) – Additional arguments for template

Returns:

DynamicPipelineTemplate instance ready for pipeline generation

Raises:

PipelineAPIError – If template creation fails

Return type:

DynamicPipelineTemplate

get_last_template()

Get the last template used during compilation.

This template will have its pipeline_metadata populated from the generation process. Use this method to get access to a template that has gone through the complete pipeline generation process, particularly useful for execution document generation.

Returns:

The last template used in compilation, or None if no compilation has occurred

Return type:

Optional[‘DynamicPipelineTemplate’]

get_supported_step_types()

Get list of supported step types.

Returns:

List of supported step type names

Return type:

list

preview_resolution(dag)

Preview how DAG nodes will be resolved to configs and builders.

Returns a detailed preview showing: - Node → Configuration mappings - Configuration → Step Builder mappings - Detected step types and dependencies - Potential issues or ambiguities

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file

Returns:

ResolutionPreview with detailed resolution information

Return type:

ResolutionPreview

validate_config_file()

Validate the configuration file structure.

Returns:

Dictionary with validation results

Return type:

Dict[str, Any]

validate_dag_compatibility(dag)

Validate that DAG nodes have corresponding configurations.

Returns detailed validation results including: - Missing configurations - Unresolvable step builders - Configuration validation errors - Dependency resolution issues

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file

Returns:

ValidationResult with detailed validation information

Return type:

ValidationResult

StepConfigResolver

alias of StepConfigResolverAdapter

class ResolutionPreview(*, node_config_map, config_builder_map, resolution_confidence, ambiguous_resolutions=<factory>, recommendations=<factory>)

Bases: BaseModel

Preview of how DAG nodes will be resolved.

display()

Display-friendly resolution preview.

model_config: ClassVar[ConfigDict] = {}

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

node_config_map: Dict[str, str]

config_builder_map: Dict[str, str]

resolution_confidence: Dict[str, float]

ambiguous_resolutions: List[str]

recommendations: List[str]

class ConversionReport(*, pipeline_name, steps, resolution_details, avg_confidence, warnings=<factory>, metadata=<factory>)

Bases: BaseModel

Report generated after successful pipeline conversion.

detailed_report()

Detailed conversion report.

model_config: ClassVar[ConfigDict] = {}

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

summary()

Summary of conversion results.

pipeline_name: str

steps: List[str]

resolution_details: Dict[str, Dict[str, Any]]

avg_confidence: float

warnings: List[str]

metadata: Dict[str, Any]

class ValidationEngine

Bases: object

Engine for validating DAG-config compatibility.

validate_dag_compatibility(dag_nodes, available_configs, config_map, builder_registry, metadata=None)

Validate DAG-config compatibility.

Parameters:

• dag_nodes (List[str]) – List of DAG node names

• available_configs (Dict[str, Any]) – Available configuration instances

• config_map (Dict[str, Any]) – Resolved node-to-config mapping

• builder_registry (Dict[str, Any]) – Available step builders

• metadata (Dict[str, Any] | None) – Optional pipeline metadata; its config_types map (node -> config class) records USER-AUTHORED node→config bindings. The node-vs-config cross-check honors it, so a deliberately off-convention node name explicitly mapped there is not flagged as a misresolution.

Returns:

ValidationResult with detailed validation information

Return type:

ValidationResult

generate_random_word(length=4)

Generate a random word of specified length.

Parameters:

length (int) – Length of the random word

Returns:

Random string of specified length

Return type:

str

validate_pipeline_name(name)

Validate that a pipeline name conforms to SageMaker constraints.

Parameters:

name (str) – The pipeline name to validate

Returns:

True if the name is valid, False otherwise

Return type:

bool

sanitize_pipeline_name(name)

Sanitize a pipeline name to conform to SageMaker constraints.

This function: 1. Replaces dots with hyphens 2. Replaces underscores with hyphens 3. Removes any other special characters 4. Ensures the name starts with an alphanumeric character 5. Ensures the name ends with an alphanumeric character

Parameters:

name (str) – The pipeline name to sanitize

Returns:

A sanitized version of the name that conforms to SageMaker constraints

Return type:

str

generate_pipeline_name(base_name, version='1.0')

Generate a valid pipeline name with the format: {base_name}-{random_word}-{version}-pipeline

This function ensures the generated name conforms to SageMaker constraints by sanitizing it before returning.

Parameters:

• base_name (str) – Base name for the pipeline

• version (str) – Version string to include in the name

Returns:

A string with the generated pipeline name that passes SageMaker validation

Return type:

str

exception PipelineAPIError

Bases: Exception

Base exception for all Pipeline API errors.

exception ConfigurationError(message, missing_configs=None, available_configs=None)

Bases: PipelineAPIError

Raised when configuration-related errors occur.

exception AmbiguityError(message, node_name=None, candidates=None)

Bases: PipelineAPIError

Raised when multiple configurations could match a DAG node.

exception ValidationError(message, validation_errors=None)

Bases: PipelineAPIError

Raised when DAG-config validation fails.

exception ResolutionError(message, failed_nodes=None, suggestions=None)

Bases: PipelineAPIError

Raised when DAG node resolution fails.

merge_and_save_configs(config_list, output_file, processing_step_config_base_class=None, workspace_dirs=None)

Merge and save multiple configs to a single JSON file.

Uses UnifiedConfigManager for streamlined processing with workspace awareness.

Parameters:

• config_list (List[Any]) – List of configuration objects to merge and save

• output_file (str) – Path to the output JSON file

• processing_step_config_base_class (type | None) – Optional base class to identify processing step configs

• workspace_dirs (List[str] | None) – Optional list of workspace directories for step catalog integration

Returns:

The categorized configuration structure

Return type:

dict

Raises:

• ValueError – If config_list is empty or contains invalid configs

• IOError – If there’s an issue writing to the output file

• TypeError – If configs are not serializable

load_configs(input_file, config_classes=None, workspace_dirs=None)

Load multiple configs from a JSON file.

Uses UnifiedConfigManager for streamlined processing with workspace awareness.

Parameters:

• input_file (str) – Path to the input JSON file

• config_classes (Dict[str, Type] | None) – Optional dictionary mapping class names to class types If not provided, UnifiedConfigManager discovery will be used

• workspace_dirs (List[str] | None) – Optional list of workspace directories for step catalog integration

Returns:

A dictionary with the following structure:

{

“shared”: {shared_field1: value1, …}, “specific”: {

”StepName1”: {specific_field1: value1, …}, “StepName2”: {specific_field2: value2, …}, …

}

}

Return type:

dict

Raises:

• FileNotFoundError – If the input file doesn’t exist

• json.JSONDecodeError – If the input file is not valid JSON

• KeyError – If required keys are missing from the file

• TypeError – If deserialization fails due to type mismatches

serialize_config(config)

Serialize a configuration object to a JSON-serializable dictionary.

This function serializes a configuration object, preserving its type information and special fields. It embeds metadata including the step name derived from attributes like ‘job_type’, ‘data_type’, and ‘mode’.

Parameters:

config (Any) – The configuration object to serialize

Returns:

A serialized representation of the config

Return type:

dict

Raises:

TypeError – If the config is not serializable

deserialize_config(data, config_classes=None)

Deserialize a dictionary back into a configuration object.

This function deserializes a dictionary into a configuration object based on type information embedded in the dictionary. If the dictionary contains the __model_type__ field, it will attempt to reconstruct the original object type using the step catalog system.

Parameters:

• data (Dict[str, Any]) – The serialized dictionary

• config_classes (Dict[str, Type] | None) – Optional dictionary mapping class names to class types If not provided, all classes registered with ConfigClassStore will be used

Returns:

The deserialized configuration object

Return type:

Any

Raises:

TypeError – If the data cannot be deserialized to the specified type

ConfigClassStore

alias of ConfigClassStoreAdapter

register_config_class(cls)

Register a configuration class with the ConfigClassStore.

This is a convenient alias for ConfigClassStore.register decorator.

Parameters:

cls (Any) – The class to register

Returns:

The class, allowing this to be used as a decorator

Return type:

Any

class PropertyReference(*, step_name, output_spec)

Bases: BaseModel

Lazy evaluation reference bridging definition-time and runtime.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'validate_assignment': True}

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

to_runtime_property(step_instances)

Create an actual SageMaker property reference using step instances.

This method navigates the property path to create a proper SageMaker Properties object that can be used at runtime.

Parameters:

step_instances (Dict[str, Any]) – Dictionary mapping step names to step instances

Returns:

SageMaker Properties object for the referenced property

Raises:

• ValueError – If the step is not found or property path is invalid

• AttributeError – If any part of the property path is invalid

Return type:

Any

to_sagemaker_property()

Convert to SageMaker Properties dictionary format at pipeline definition time.

classmethod validate_step_name(v)

Validate step name is not empty.

step_name: str

output_spec: OutputDecl

class SpecificationRegistry(context_name='default')

Bases: object

Context-aware registry for managing step specifications with isolation.

find_compatible_outputs(dependency_spec)

Find outputs compatible with a dependency specification.

get_specification(step_name)

Get specification by step name.

get_specifications_by_type(step_type)

Get all specifications of a given step type.

list_step_names()

Get list of all registered step names.

list_step_types()

Get list of all registered step types.

register(step_name, specification)

Register a step specification.

class RegistryManager(workspace_context=None)

Bases: object

Manager for context-scoped registries with complete isolation and workspace awareness.

This enhanced registry manager now supports: 1. Workspace-aware registry resolution 2. Integration with hybrid registry system 3. Backward compatibility with existing code 4. Thread-local workspace context management

clear_all_contexts()

Clear all registries.

clear_context(context_name)

Clear the registry for a specific context.

Parameters:

context_name (str) – Name of the context to clear

Returns:

True if the registry was cleared, False if it didn’t exist

Return type:

bool

get_context_stats()

Get statistics for all contexts.

Returns:

Dictionary mapping context names to their statistics

Return type:

Dict[str, Dict[str, int]]

get_registry(context_name='default', create_if_missing=True)

Get the registry for a specific context with workspace awareness.

This enhanced method now supports: 1. Workspace-aware context naming 2. Integration with hybrid registry system 3. Automatic registry population from hybrid sources 4. Backward compatibility with existing code

Parameters:

• context_name (str) – Name of the context (e.g., pipeline name, environment)

• create_if_missing (bool) – Whether to create a new registry if one doesn’t exist

Returns:

Context-specific registry or None if not found and create_if_missing is False

Return type:

SpecificationRegistry

list_contexts()

Get list of all registered context names.

Returns:

List of context names with registries

Return type:

List[str]

get_registry(manager=None, context_name='default')

Get the registry for a specific context.

Parameters:

• manager (RegistryManager | None) – Registry manager instance

• context_name (str) – Name of the context (e.g., pipeline name, environment)

Returns:

Context-specific registry

Return type:

SpecificationRegistry

get_pipeline_registry(manager, pipeline_name)

Get registry for a pipeline (backward compatibility).

Parameters:

• manager (RegistryManager) – Registry manager instance

• pipeline_name (str) – Name of the pipeline

Returns:

Pipeline-specific registry

Return type:

SpecificationRegistry

get_default_registry(manager)

Get the default registry (backward compatibility).

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

Default registry

Return type:

SpecificationRegistry

integrate_with_pipeline_builder(pipeline_builder_cls, manager=None)

Decorator to integrate context-scoped registries with a pipeline builder class.

This decorator modifies a pipeline builder class to use context-scoped registries.

Parameters:

• pipeline_builder_cls (Any) – Pipeline builder class to modify

• manager (RegistryManager | None) – Registry manager instance (if None, a new instance will be created)

Returns:

Modified pipeline builder class

Return type:

Any

list_contexts(manager)

Get list of all registered context names.

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

List of context names with registries

Return type:

List[str]

clear_context(manager, context_name)

Clear the registry for a specific context.

Parameters:

• manager (RegistryManager) – Registry manager instance

• context_name (str) – Name of the context to clear

Returns:

True if the registry was cleared, False if it didn’t exist

Return type:

bool

get_context_stats(manager)

Get statistics for all contexts.

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

Dictionary mapping context names to their statistics

Return type:

Dict[str, Dict[str, int]]

class UnifiedDependencyResolver(registry, semantic_matcher)

Bases: object

Intelligent dependency resolver using declarative specifications.

clear_cache()

Clear the resolution cache.

get_resolution_report(available_steps)

Generate a detailed resolution report for debugging.

Parameters:

available_steps (List[str]) – List of available step names

Returns:

Detailed report of resolution process

Return type:

Dict[str, Any]

register_specification(step_name, spec)

Register a step specification with the resolver.

resolve_all_dependencies(available_steps)

Resolve dependencies for all registered steps.

Parameters:

available_steps (List[str]) – List of step names that are available in the pipeline

Returns:

Dictionary mapping step names to their resolved dependencies

Return type:

Dict[str, Dict[str, PropertyReference]]

resolve_step_dependencies(consumer_step, available_steps)

Resolve dependencies for a single step.

Parameters:

• consumer_step (str) – Name of the step whose dependencies to resolve

• available_steps (List[str]) – List of available step names

Returns:

Dictionary mapping dependency names to property references

Return type:

Dict[str, PropertyReference]

resolve_with_scoring(consumer_step, available_steps)

Resolve dependencies with detailed compatibility scoring.

Parameters:

• consumer_step (str) – Name of the step whose dependencies to resolve

• available_steps (List[str]) – List of available step names

Returns:

Dictionary with resolved dependencies and detailed scoring information

Return type:

Dict[str, Any]

exception DependencyResolutionError

Bases: Exception

Raised when dependencies cannot be resolved.

create_dependency_resolver(registry=None, semantic_matcher=None)

Create a properly configured dependency resolver.

Parameters:

• registry (SpecificationRegistry | None) – Optional specification registry. If None, creates a new one.

• semantic_matcher (SemanticMatcher | None) – Optional semantic matcher. If None, creates a new one.

Returns:

Configured UnifiedDependencyResolver instance

Return type:

UnifiedDependencyResolver

class SemanticMatcher

Bases: object

Semantic similarity matching for dependency resolution.

calculate_similarity(name1, name2)

Calculate semantic similarity between two names.

Parameters:

• name1 (str) – First name to compare

• name2 (str) – Second name to compare

Returns:

Similarity score between 0.0 and 1.0

Return type:

float

calculate_similarity_with_aliases(name, output_spec)

Calculate semantic similarity between a name and an output specification, considering both logical_name and all aliases.

Parameters:

• name (str) – The name to compare (typically the dependency’s logical_name)

• output_spec (Any) – OutputSpec with logical_name and potential aliases

Returns:

The highest similarity score (0.0 to 1.0) between name and any name in output_spec

Return type:

float

explain_similarity(name1, name2)

Provide detailed explanation of similarity calculation.

Parameters:

• name1 (str) – First name to compare

• name2 (str) – Second name to compare

Returns:

Dictionary with detailed similarity breakdown

Return type:

Dict[str, Any]

find_best_matches(target_name, candidate_names, threshold=0.5)

Find the best matching names from a list of candidates.

Parameters:

• target_name (str) – Name to match against

• candidate_names (List[str]) – List of candidate names

• threshold (float) – Minimum similarity threshold

Returns:

List of (name, score) tuples sorted by score (highest first)

Return type:

List[Tuple[str, float]]

create_pipeline_components(context_name=None)

Create all necessary pipeline components with proper dependencies.

Modules

assembler	Pipeline assembler module.
base	Core base classes for the cursus framework.
compiler	Pipeline compiler module.
config_fields	Configuration Field Manager Package.
deps	Pipeline Dependencies module - Declarative dependency management for SageMaker pipelines.
utils	Core utilities for cursus.