cursus.core.assembler

Pipeline assembler module.

This module provides classes for assembling and building SageMaker pipelines from DAG specifications and configurations.

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

pipeline_metadata: Dict[str, Any]

Modules

pipeline_assembler	Pipeline assembler that builds pipelines from a DAG structure and step builders.
pipeline_template_base	Base class for pipeline templates.