cursus.core.compiler¶

Pipeline compiler module.

This module provides high-level interfaces for compiling PipelineDAG structures directly into executable SageMaker pipelines without requiring custom template classes.

compile_dag_to_pipeline(dag=None, dag_path=None, config_path=None, sagemaker_session=None, role=None, pipeline_name=None, **kwargs)[source]¶

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

compile_single_node_to_pipeline(dag, config_path, target_node, manual_inputs, sagemaker_session=None, role=None, pipeline_name=None, validate_inputs=True, config_map=None, pipeline_parameters=None, **kwargs)[source]¶

Compile a single-node pipeline with manual input overrides.

IMPROVED API: config_map is now optional and will be automatically loaded from config_path if not provided, matching the API consistency of compile_dag_to_pipeline().

This function creates a SingleNodeCompiler and delegates to its compile method, providing a simple one-line API for single-node execution.

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file
config_path (str) – Path to configuration JSON file
target_node (str) – Name of node to execute in isolation
manual_inputs (Dict[str, str]) – Dict mapping logical input names to S3 URIs Example: {“input_path”: “s3://bucket/previous-run/output/”}
sagemaker_session (Any | None) – SageMaker session for pipeline execution
role (str | None) – IAM role ARN for SageMaker permissions
pipeline_name (str | None) – Optional custom pipeline name Default: “{target_node}-isolated”
validate_inputs (bool) – Whether to validate inputs before compilation Checks: S3 URI format, node existence (default: True)
config_map (Dict | None) – Optional pre-loaded config map for advanced use cases If None (default), automatically loaded from config_path
pipeline_parameters (List[str | ParameterString] | None) – Pipeline parameters to pass to compiler. If None, uses default parameters (EXECUTION_S3_PREFIX, KMS_KEY, etc.)
**kwargs (Any) – Additional arguments passed to SingleNodeCompiler

Returns:

Single-node Pipeline ready for execution via pipeline.start()

Raises:

ValueError – If validation fails or node not found
FileNotFoundError – If config_path doesn’t exist

Return type:

Any

Example (Simple - Config Auto-Loaded):

>>> # After a 5-hour run where preprocess succeeded but train failed
>>> manual_inputs = {
...     "input_path": "s3://my-bucket/run-123/preprocess/output/"
... }
>>>
>>> pipeline = compile_single_node_to_pipeline(
...     dag=my_dag,
...     config_path="configs/pipeline.json",  # ✓ Auto-loads config
...     target_node="train",
...     manual_inputs=manual_inputs,
...     sagemaker_session=session,
...     role=role
... )
>>>
>>> # Execute just the train step - no 5-hour wait!
>>> execution = pipeline.start()

Example (Advanced - Pre-Loaded Config):

>>> # For performance optimization when config already loaded
>>> config_map = load_configs("configs/pipeline.json")
>>>
>>> pipeline = compile_single_node_to_pipeline(
...     dag=my_dag,
...     config_path="configs/pipeline.json",
...     target_node="train",
...     manual_inputs={"input_path": "s3://..."},
...     config_map=config_map,  # Pass pre-loaded config
...     sagemaker_session=session,
...     role=role
... )

Benefits:

28% time savings on debugging failed pipelines
33% cost savings by avoiding redundant computation
3× faster iteration cycles during development
API consistency with compile_dag_to_pipeline()

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)[source]¶

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()[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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()[source]¶

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()[source]¶

Get list of supported step types.

Returns:: List of supported step type names
Return type:: list

preview_resolution(dag)[source]¶

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()[source]¶

Validate the configuration file structure.

Returns:: Dictionary with validation results
Return type:: Dict[str, Any]

validate_dag_compatibility(dag)[source]¶

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

class SingleNodeCompiler(config_path, sagemaker_session=None, role=None, step_catalog=None, pipeline_parameters=None, project_root=None, anchor_file=None, **kwargs)[source]¶

Bases: object

Specialized compiler for single-node pipeline execution.

Enables rapid debugging by creating isolated pipelines containing just one node, with manual input overrides bypassing normal dependency resolution.

Example

>>> compiler = SingleNodeCompiler(
...     config_path="configs/pipeline.json",
...     sagemaker_session=session,
...     role=role
... )
>>>
>>> # Validate before execution
>>> validation = compiler.validate_node_and_inputs(
...     dag=dag,
...     target_node="train",
...     manual_inputs={"input_path": "s3://bucket/data/"}
... )
>>>
>>> if validation.is_valid:
...     pipeline = compiler.compile(
...         dag=dag,
...         target_node="train",
...         manual_inputs={"input_path": "s3://bucket/data/"}
...     )

compile(dag, target_node, manual_inputs, pipeline_name=None, validate_inputs=True, config_map=None, **assembler_kwargs)[source]¶

Compile single-node pipeline with manual inputs.

IMPROVED: config_map is now optional! If not provided, it will be automatically loaded from the config_path using the same mechanism as compile_dag_to_pipeline().

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file
target_node (str) – Name of node to execute
manual_inputs (Dict[str, str]) – Manual input paths (logical_name -> s3_uri)
pipeline_name (str | None) – Optional pipeline name
validate_inputs (bool) – Whether to validate inputs (default: True)
config_map (Dict | None) – Optional pre-loaded config map (auto-loaded if None)
**assembler_kwargs (Any) – Additional arguments for PipelineAssembler

Returns:

Single-node Pipeline ready for execution

Raises:

ValueError – If validation fails or compilation errors occur
FileNotFoundError – If config file not found (when auto-loading)

Return type:

Any

preview_execution(dag, target_node, manual_inputs)[source]¶

Preview execution without creating pipeline.

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file
target_node (str) – Name of node to preview
manual_inputs (Dict[str, str]) – Manual input paths

Returns:

ExecutionPreview with detailed execution information

Return type:

ExecutionPreview

validate_node_and_inputs(dag, target_node, manual_inputs)[source]¶

Validate target node and manual inputs before execution.

Parameters:

dag (PipelineDAG | str) – PipelineDAG instance or path to serialized DAG file
target_node (str) – Name of node to validate
manual_inputs (Dict[str, str]) – Manual input paths to validate

Returns:

ValidationResult with detailed validation information

Return type:

ValidationResult

StepConfigResolver¶: alias of StepConfigResolverAdapter

class ValidationResult(*, is_valid, missing_configs=<factory>, unresolvable_builders=<factory>, config_errors=<factory>, dependency_issues=<factory>, warnings=<factory>)[source]¶

Bases: BaseModel

Result of DAG-config compatibility validation.

detailed_report()[source]¶

Detailed validation report with recommendations.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

summary()[source]¶

Human-readable validation summary.

is_valid: bool¶

missing_configs: List[str]¶

unresolvable_builders: List[str]¶

config_errors: Dict[str, List[str]]¶

dependency_issues: List[str]¶

warnings: List[str]¶

SingleNodeValidationResult¶: alias of ValidationResult

class ExecutionPreview(target_node, step_type, config_type, input_mappings=<factory>, missing_required_inputs=<factory>, missing_optional_inputs=<factory>, output_paths=<factory>, estimated_instance_type='Unknown', estimated_duration=None, warnings=<factory>)[source]¶

Bases: object

Preview of single-node execution.

display()[source]¶

Generate formatted display string.

estimated_duration: str | None = None¶

estimated_instance_type: str = 'Unknown'¶

target_node: str¶

step_type: str¶

config_type: str¶

input_mappings: Dict[str, str]¶

missing_required_inputs: List[str]¶

missing_optional_inputs: List[str]¶

output_paths: Dict[str, str]¶

warnings: List[str]¶

class ResolutionPreview(*, node_config_map, config_builder_map, resolution_confidence, ambiguous_resolutions=<factory>, recommendations=<factory>)[source]¶

Bases: BaseModel

Preview of how DAG nodes will be resolved.

display()[source]¶

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>)[source]¶

Bases: BaseModel

Report generated after successful pipeline conversion.

detailed_report()[source]¶

Detailed conversion report.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

summary()[source]¶

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[source]¶

Bases: object

Engine for validating DAG-config compatibility.

validate_dag_compatibility(dag_nodes, available_configs, config_map, builder_registry, metadata=None)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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')[source]¶

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[source]¶

Bases: Exception

Base exception for all Pipeline API errors.

exception ConfigurationError(message, missing_configs=None, available_configs=None)[source]¶

Bases: PipelineAPIError

Raised when configuration-related errors occur.

exception AmbiguityError(message, node_name=None, candidates=None)[source]¶

Bases: PipelineAPIError

Raised when multiple configurations could match a DAG node.

exception ValidationError(message, validation_errors=None)[source]¶

Bases: PipelineAPIError

Raised when DAG-config validation fails.

exception ResolutionError(message, failed_nodes=None, suggestions=None)[source]¶

Bases: PipelineAPIError

Raised when DAG node resolution fails.

Modules

`dag_compiler`	DAG Compiler for the Pipeline API.
`dynamic_template`	Dynamic Pipeline Template for the Pipeline API.
`exceptions`	Exception classes for the Pipeline API.
`name_generator`	Name generator utilities for pipeline naming.
`single_node_compiler`	Single node execution compiler for debugging and rapid iteration.
`validation`	Validation and preview classes for the Pipeline API.