cursus.core.assembler.pipeline_assembler

Pipeline assembler that builds pipelines from a DAG structure and step builders.

This assembler leverages the specification-based dependency resolution system to intelligently connect steps and build complete SageMaker pipelines.

safe_value_for_logging(value)

Safely format a value for logging, handling Pipeline variables appropriately.

Parameters:

value (Any) – Any value that might be a Pipeline variable

Returns:

A string representation safe for logging

Return type:

str

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.

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

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.