cursus.core.compiler.dag_compiler

DAG Compiler for the Pipeline API.

This module provides the main API functions for compiling PipelineDAG structures into executable SageMaker pipelines.

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.

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

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

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_supported_step_types()

Get list of supported step types.

Returns:

List of supported step type names

Return type:

list

validate_config_file()

Validate the configuration file structure.

Returns:

Dictionary with validation results

Return type:

Dict[str, Any]

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’]

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