cursus

Cursus: Automatic SageMaker Pipeline Generation

Transform pipeline graphs into production-ready SageMaker pipelines automatically. An intelligent pipeline generation system that automatically creates complete SageMaker pipelines from user-provided pipeline graphs with intelligent dependency resolution and configuration management.

Key Features: - 🎯 Graph-to-Pipeline Automation: Automatically generate complete SageMaker pipelines - ⚡ 10x Faster Development: Minutes to working pipeline vs. weeks of manual configuration - 🧠 Intelligent Dependency Resolution: Automatic step connections and data flow - 🛡️ Production Ready: Built-in quality gates and validation - 📈 Proven Results: 60% average code reduction across pipeline components

Basic Usage:

>>> import cursus
>>> pipeline = cursus.compile_dag(my_dag)

>>> from cursus import PipelineDAGCompiler
>>> compiler = PipelineDAGCompiler()
>>> pipeline = compiler.compile(my_dag, pipeline_name="fraud-detection")

Advanced Usage:

>>> from cursus import PipelineDAG, compile_dag_to_pipeline
>>>
>>> dag = PipelineDAG()
>>> # ... build your DAG
>>> pipeline = compile_dag_to_pipeline(dag, config_path="config.yaml")

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

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

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

Create a SageMaker pipeline from a DAG specification.

This is a convenience function that combines DAG compilation and pipeline creation in a single call with sensible defaults.

Parameters:

• dag – PipelineDAG instance or DAG specification

• pipeline_name – Optional name for the pipeline

• **kwargs – Additional arguments passed to the compiler

Returns:

SageMaker Pipeline instance ready for execution

Example

>>> dag = PipelineDAG()
>>> # ... configure your DAG
>>> pipeline = create_pipeline_from_dag(dag, "my-ml-pipeline")
>>> pipeline.start()

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

class PipelineDAG(nodes=None, edges=None, strict=False)

Bases: object

Represents a pipeline topology as a directed acyclic graph (DAG). Each node is a step name; edges define dependencies.

Node declaration vs. auto-creation: a node is “declared” when it is passed in the nodes= constructor arg or added via add_node(). By default add_edge() auto-creates any endpoint that was not declared (lenient mode) — convenient, but it means a single typo in an edge name (add_edge("A", "TabularPreprocessing_traning")) silently spawns a phantom, unconfigured node and orphans the real one, and construction never raises. Two guards exist:

• validate_node_declarations() reports every edge endpoint that was never declared (always available, non-fatal) — call it to surface likely typos / forgotten add_node.

• strict=True turns the same condition into an immediate ValueError at add_edge (and constructor) time, so undeclared endpoints can never enter the graph.

add_edge(src, dst)

Add a directed edge from src to dst.

Lenient (default): auto-creates either endpoint if it is not yet a node — but does NOT mark it declared, so validate_node_declarations will still surface it as a likely typo. strict=True: raises ValueError if either endpoint was never declared via add_node.

add_node(node)

Add a single node to the DAG. This DECLARES the node (see class docstring).

get_dependencies(node)

Return immediate dependencies (parents) of a node.

topological_sort()

Return nodes in topological order.

Tolerates edges whose endpoints are not in nodes (dangling edges) by ignoring the unknown endpoints here rather than raising an opaque KeyError — structural problems like dangling edges are surfaced with clear messages by the serializer’s validation layer.

validate_node_declarations()

Return edge endpoints that were never explicitly declared via add_node / the nodes= arg.

Because add_edge auto-creates missing endpoints (lenient mode), a typo in an edge name silently produces a phantom, unconfigured node — and the serializer’s dangling-edge check cannot catch it, since add_edge has already promoted the typo into nodes. This is the only reliable detector: it compares every edge endpoint against the DECLARED set. An empty list means every edge endpoint was declared; any member is a likely typo or a forgotten add_node. Non-fatal — strict=True turns the same condition into a raise at add_edge time.

class DynamicPipelineTemplate(dag, config_path, config_resolver=None, step_catalog=None, skip_validation=False, pipeline_parameters=None, **kwargs)

Bases: PipelineTemplateBase

Dynamic pipeline template that works with any PipelineDAG.

This template automatically implements the abstract methods of PipelineTemplateBase by using intelligent resolution mechanisms to map DAG nodes to configurations and step builders.

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

get_execution_order()

Get the topological execution order of steps.

Returns:

List of step names in execution order

Return type:

list

get_resolution_preview()

Get a preview of how DAG nodes will be resolved.

Returns:

Dictionary with resolution preview information

Return type:

Dict[str, Any]

get_step_catalog_stats()

Get statistics about the step catalog.

Returns:

Dictionary with step catalog statistics

Return type:

Dict[str, Any]

get_step_dependencies()

Get the dependencies for each step based on the DAG.

Returns:

Dictionary mapping step names to their dependencies

Return type:

Dict[str, list]

validate_before_build()

Validate the configuration before building the pipeline.

Returns:

True if validation passes, False otherwise

Return type:

bool

Modules

api	Cursus API module.
cli	Command-line interface for the Cursus package.
core	Cursus Core module.
mcp	cursus.mcp — Cursus as an agentic toolset.
mods	Cursus modules package.
pipeline_catalog	Pipeline Catalog — queryable DAG store with common pipeline builders.
processing	Cursus Processing Module
registry	Enhanced Pipeline Registry Module with Hybrid Registry Support.
step_catalog	Unified Step Catalog System
steps	Pipeline Steps Module.
validation	Cursus Validation Framework