cursus.pipeline_catalog

Pipeline Catalog — queryable DAG store with common pipeline builders.

Usage:

# Quick compile (SAIS notebook): from cursus.pipeline_catalog import build_and_compile pipeline, report = build_and_compile(dag_path=”dag.json”, config_path=”config.json”, …)

# Generate MODS class (for MODS Lambda): from cursus.pipeline_catalog import build_mods_pipeline MyPipeline = build_mods_pipeline(author=”…”, version=”…”, dag_path=”…”, config_path=”…”)

# Search catalog: from cursus.pipeline_catalog import search_dags results = search_dags(features=[“bedrock”, “training”], framework=”pytorch”)

build_and_compile(dag_path, config_path, sagemaker_session=None, role=None, project_root=None)

Build and compile a pipeline from DAG + config paths. No class needed.

Parameters:

• dag_path (str) – Path to .dag.json file

• config_path (str) – Path to config JSON

• sagemaker_session (Session | None) – SageMaker session

• role (str | None) – IAM role ARN

• project_root (str | Path | None) – Optional explicit project folder for docker source_dir resolution (the caller hook). When omitted, the compiler infers it from config_path.

Returns:

(Pipeline, CompilationReport)

Return type:

Tuple[Pipeline, any]

build_mods_pipeline(author, version, description, dag_path, config_path, class_name=None)

Generate a @MODSTemplate-decorated pipeline class from declarative config.

The generated class has the standard MODS interface: - __init__(sagemaker_session, execution_role, regional_alias) - generate_pipeline() -> Pipeline

Parameters:

• author (str) – Pipeline author alias

• version (str) – Pipeline version string

• description (str) – Pipeline description

• dag_path (str) – Relative path to DAG JSON (relative to the calling module’s directory)

• config_path (str) – Relative path to config JSON

• class_name (str | None) – Optional class name (default: derived from description)

Returns:

A @MODSTemplate decorated class ready for MODS Lambda

Return type:

Type

recommend_dag(framework=None, features=None, task_type=None, complexity=None, max_results=5)

Recommend DAGs based on requirements. Returns ranked list.

Scoring (0-1):

• Feature overlap: |required ∩ dag_features| / |required| (weight: 0.5)

• Framework match: 1.0 if exact match (weight: 0.25)

• Task type match: 1.0 if substring match (weight: 0.15)

• Complexity match: 1.0 if exact, 0.5 if adjacent (weight: 0.1)

Parameters:

• framework (str | None) – Required framework (pytorch, xgboost, lightgbm, etc.)

• features (List[str] | None) – Required features (e.g., [“training”, “bedrock”, “edx_uploading”])

• task_type (str | None) – Task type keyword (e.g., “incremental”, “end_to_end”)

• complexity (str | None) – Desired complexity (simple, standard, advanced, comprehensive)

• max_results (int) – Maximum number of results to return

Returns:

List of dicts with ‘id’, ‘score’, ‘reasoning’, and all DAG metadata

Return type:

List[Dict[str, Any]]

auto_select_dag(framework=None, features=None, task_type=None, min_score=0.6)

Auto-select the best matching DAG. Returns None if no good match.

Parameters:

• framework (str | None) – Required framework

• features (List[str] | None) – Required features

• task_type (str | None) – Task type keyword

• min_score (float) – Minimum score threshold (0-1)

Returns:

Tuple of (dag_id, PipelineDAG, score) or None

Return type:

Tuple[str, PipelineDAG, float] | None

create_pipeline(dag_id=None, dag_path=None, config_path=None, sagemaker_session=None, role=None)

Create a pipeline from a DAG + config.

Parameters:

• dag_id (str | None) – Shared DAG ID from catalog (e.g., “bedrock_pytorch_incremental_edx”)

• dag_path (str | None) – Path to a .dag.json file (alternative to dag_id)

• config_path (str) – Path to pipeline config JSON

• sagemaker_session (Session | None) – SageMaker session

• role (str | None) – IAM role ARN

Returns:

Tuple of (Pipeline, CompilationReport)

Return type:

Tuple[Pipeline, any]

list_available_pipelines()

List all available shared DAG IDs.

load_shared_dag(dag_id)

Load a shared DAG by ID from the JSON catalog.

Parameters:

dag_id (str) – DAG identifier (e.g., “bedrock_pytorch_incremental_edx”)

Returns:

PipelineDAG ready for compilation

Return type:

PipelineDAG

search_dags(features=None, framework=None)

Search DAGs by features and/or framework.

Parameters:

• features (List[str] | None) – List of required features (e.g., [“training”, “bedrock”, “edx_uploading”])

• framework (str | None) – Framework filter (e.g., “pytorch”)

Returns:

List of matching DAG entries, sorted by feature overlap

Return type:

List[Dict[str, Any]]

get_all_shared_dags()

Get metadata for all available shared DAGs from the catalog index.

Returns:

Dict mapping DAG id to metadata dict

Return type:

Dict[str, Dict[str, Any]]

get_catalog_index()

Load the catalog index.

Modules

core	Pipeline Catalog Core — factory, builders, and router.
mods_pipelines	Pipeline Catalog - MODS Pipelines (namespace stub)
shared_dags	Shared DAG Definitions for Pipeline Catalog