cursus.step_catalog.step_catalog

Unified Step Catalog - Single class addressing all US1-US5 requirements.

This module implements the core StepCatalog class that consolidates 16+ fragmented discovery mechanisms into a single, efficient system with O(1) lookups and intelligent component discovery across multiple workspaces.

set_default_workspace_dirs(dirs)

Push process-level default step-pack dirs for bare StepCatalog() construction.

Called by the DAG compiler when a plugin pack is resolved, so validate/author/exec-doc sites that build a bare catalog also see the plugin steps. Pass None/empty to clear. Additive only — never suppresses package discovery.

get_default_workspace_dirs()

Return the current process-level default step-pack dirs (may be empty).

class StepCatalog(workspace_dirs=None)

Bases: object

Unified step catalog addressing all validated user stories (US1-US5).

This single class consolidates the functionality of 16+ discovery systems while maintaining simple, efficient O(1) lookups through dictionary-based indexing.

PHASE 1 ENHANCEMENT: Now includes StepBuilderRegistry functionality: - Config-to-builder resolution - Legacy alias support - Pipeline construction interface - Enhanced registry integration

LEGACY_ALIASES = {'MIMSPackaging': 'Package', 'MIMSPayload': 'Payload', 'ModelRegistration': 'Registration', 'PytorchModel': 'PyTorchModel', 'PytorchTraining': 'PyTorchTraining'}

metrics: Dict[str, Any]

get_step_info(step_name, job_type=None)

Get complete information about a step, optionally with job_type variant.

Parameters:

• step_name (str) – Name of the step to retrieve

• job_type (str | None) – Optional job type variant (e.g., ‘training’, ‘validation’)

Returns:

StepInfo object with complete step information, or None if not found

Return type:

StepInfo | None

find_step_by_component(component_path)

Find step name from any component file.

Parameters:

component_path (str) – Path to a component file

Returns:

Step name that owns the component, or None if not found

Return type:

str | None

list_available_steps(workspace_id=None, job_type=None)

List all available concrete pipeline steps with deduplication.

Excludes base configuration steps (‘Base’, ‘Processing’) and applies canonical name deduplication following standardization rules.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of concrete pipeline step names (PascalCase, canonical)

Return type:

List[str]

list_steps_with_specs(workspace_id=None, job_type=None)

List all steps that have specification components.

This method filters available steps to only return those that have specification file components, which is useful for validation frameworks that need to work specifically with steps that have specifications.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of step names that have specification components

Return type:

List[str]

list_steps_with_scripts(workspace_id=None, job_type=None)

List all steps that have script components.

This method filters available steps to only return those that have script file components, which is useful for alignment validation and script-based testing frameworks.

Parameters:

• workspace_id (str | None) – Optional workspace filter

• job_type (str | None) – Optional job type filter

Returns:

List of step names that have script components

Return type:

List[str]

search_steps(query, job_type=None)

Search steps by name with basic fuzzy matching.

Parameters:

• query (str) – Search query string

• job_type (str | None) – Optional job type filter

Returns:

List of search results sorted by relevance

Return type:

List[StepSearchResult]

discover_config_classes(project_id=None)

Auto-discover configuration classes from core and workspace directories.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping class names to class types

Return type:

Dict[str, Type]

build_complete_config_classes(project_id=None)

Build complete mapping integrating manual registration with auto-discovery.

This addresses the TODO in the existing build_complete_config_classes() function.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Complete dictionary of config classes (manual + auto-discovered)

Return type:

Dict[str, Type]

discover_contracts_with_scripts()

DISCOVERY: Find all steps that have both contract and script components.

Returns:

List of step names that have both contract and script components

Return type:

List[str]

detect_framework(step_name)

DETECTION: Detect ML framework for a step.

Parameters:

step_name (str) – Name of the step to analyze

Returns:

Framework name (e.g., ‘xgboost’, ‘pytorch’) or None if not detected

Return type:

str | None

discover_cross_workspace_components(workspace_ids=None)

DISCOVERY: Find components across multiple workspaces.

Parameters:

workspace_ids (List[str] | None) – Optional list of workspace IDs to search (defaults to all)

Returns:

Dictionary mapping workspace IDs to lists of component identifiers

Return type:

Dict[str, List[str]]

get_builder_class_path(step_name)

Get builder class path for a step using BuilderAutoDiscovery component.

Parameters:

step_name (str) – Name of the step

Returns:

Path to builder class or None if not found

Return type:

str | None

load_builder_class(step_name)

Load builder class for a step with job type variant fallback support.

Parameters:

step_name (str) – Name of the step (may include job type variant)

Returns:

Builder class type or None if not found/loadable

Return type:

Type | None

has_builder_provider(step_name)

Whether a step has a buildable builder, WITHOUT importing/instantiating it (FZ 31e1d3g3 D4).

This is a ROUTABILITY check answered from declarative data — the registry’s sagemaker_step_type (+ the .step.yaml patterns.step_assembly) must resolve to a real PatternHandler. It does NOT call load_builder_class (which would import the builder module — and fail for an SDK-bound builder offline even though the step is perfectly routable). So it stays True for a fileless-but-routable step (the factory-shell end-state) and for the SDK steps offline, and False only for genuinely-absent or non-routable rows (Base/Lambda/unknown). Used by the DAG resolver’s component-availability check, which must not encode the deleted-builder_*.py-file assumption.

Parameters:

step_name (str) – Canonical step name (job-type variants strip to the base name).

Returns:

True iff the step’s registry row routes to a construction handler.

Return type:

bool

get_sdk_output_location(built_step, output_type=None)

Read the output S3 location(s) off a BUILT SDK-delegation step (FZ 31e1d3i).

A generic, DUCK-TYPED accessor for the MODSPredefinedProcessingStep family: any built SDK step that exposes get_output_locations (the SAIS contract) is supported — CradleDataLoadingStep returns a {DATA, SIGNATURE, METADATA} dict (or one entry when output_type is given), DataUploadingStep returns None (a SINK — output goes to Andes/EDX), and a step lacking the method (e.g. RedshiftDataLoadingStep) returns None. This REPLACES the per-step get_output_location / get_step_outputs helpers the cradle builder used to carry (so that builder becomes a pure shell). It reads a method off an ALREADY-CONSTRUCTED step instance, so it introduces NO offline SAIS import — the SAIS class was necessarily importable to build the step.

Parameters:

• built_step (Any) – A step object produced by an SDK-delegation builder’s create_step.

• output_type (str | None) – Optional SAIS output type (e.g. "DATA"); None returns all locations.

Returns:

The step’s output location(s) as the SDK reports them, or None if the step does not expose get_output_locations.

Return type:

Any

get_step_interface(step_name, job_type=None)

Load a step’s unified .step.yaml interface — the single canonical accessor.

This is THE one load function for interface data (FZ 31e1d3g3 follow-up): both load_contract_class (→ iface.contract) and load_spec_class (→ iface) are thin views over it, so the job_type-suffix fallback below is written once and every consumer inherits it.

On the first (exact) load miss, if job_type was not supplied, we resolve the node’s base step name ROBUSTLY (naming.resolve_base_step_name against the registry — validates the base against actual step names, not a hardcoded suffix list) and retry with the trailing segment as the job_type (ModelCalibration_calibration → ModelCalibration + job_type calibration; CradleDataLoading_munged → CradleDataLoading + munged). Because the base is validated against the registry, ANY suffix resolves (job_type is open now) while a base like XGBoostModel is never mis-stripped (no XGBoost step exists to strip to).

Parameters:

• step_name (str) – PascalCase step name (may be a Base_variant compound).

• job_type (str | None) – Optional job_type variant; when given, no suffix stripping is attempted.

Returns:

The validated StepInterface, or None if no interface resolves (never raises).

Return type:

Any | None

load_contract_class(step_name)

Load the contract for a step — a VIEW onto its .step.yaml interface.

Parameters:

step_name (str) – Name of the step (PascalCase, e.g., “TabularPreprocessing”)

Returns:

ContractSection (drop-in for the legacy ScriptContract), or None if not found.

Return type:

Any | None

load_spec_class(step_name)

Load the specification for a step — the .step.yaml interface itself.

The StepInterface is a drop-in for the legacy StepSpecification (exposes dependencies/outputs/…), so it IS the spec view.

Parameters:

step_name (str) – Name of the step (PascalCase)

Returns:

StepInterface (drop-in for the legacy StepSpecification), or None if not found.

Return type:

Any | None

find_specs_by_contract(contract_name)

Find all specifications that reference a specific contract.

This method enables contract-specification alignment validation by finding specifications that are associated with a given contract name.

Parameters:

contract_name (str) – Name of the contract to find specifications for

Returns:

Dictionary mapping spec names to specification instances

Return type:

Dict[str, Any]

serialize_spec(spec_instance)

Convert specification instance to dictionary format.

This method provides standardized serialization of StepSpecification objects for use in validation and alignment testing.

Parameters:

spec_instance (Any) – StepSpecification instance to serialize

Returns:

Dictionary representation of the specification

Return type:

Dict[str, Any]

load_all_specifications()

Load all specification instances from both package and workspace directories.

This method provides comprehensive specification loading for validation frameworks and dependency analysis tools. It discovers and loads all available specifications, serializing them to dictionary format for easy consumption.

Returns:

Dictionary mapping step names to serialized specification dictionaries

Return type:

Dict[str, Dict[str, Any]]

get_spec_job_type_variants(base_step_name)

Get all job type variants for a base step name from specifications.

This method discovers different job type variants (training, validation, testing, etc.) for a given base step name by examining specification file naming patterns.

Parameters:

base_step_name (str) – Base name of the step

Returns:

List of job type variants found

Return type:

List[str]

create_unified_specification(contract_name)

Create unified specification from multiple variants using smart selection.

This method delegates to SpecAutoDiscovery for smart specification handling, integrating SmartSpecificationSelector functionality: - Multi-variant specification discovery - Union of dependencies and outputs from all variants - Smart validation logic with detailed feedback - Primary specification selection (training > generic > first available)

Parameters:

contract_name (str) – Name of the contract to find specifications for

Returns:

Unified specification model with metadata

Return type:

Dict[str, Any]

validate_logical_names_smart(contract, contract_name)

Smart validation using multi-variant specification logic.

This method delegates to SpecAutoDiscovery for smart validation, implementing the core Smart Specification Selection validation: - Contract input is valid if it exists in ANY variant - Contract must cover intersection of REQUIRED dependencies - Provides detailed feedback about which variants need what

Parameters:

• contract (Dict[str, Any]) – Contract dictionary

• contract_name (str) – Name of the contract

Returns:

List of validation issues

Return type:

List[Dict[str, Any]]

serialize_contract(contract_instance)

Convert contract instance to dictionary format.

This method provides standardized serialization of ScriptContract objects for use in script-contract alignment validation.

Parameters:

contract_instance (Any) – Contract instance to serialize

Returns:

Dictionary representation of the contract

Return type:

Dict[str, Any]

find_contracts_by_entry_point(entry_point)

Find contracts that reference a specific script entry point.

Parameters:

entry_point (str) – Script entry point (e.g., “model_evaluation_xgb.py”)

Returns:

Dictionary mapping contract names to contract instances

Return type:

Dict[str, Any]

get_contract_entry_points()

Get all contract entry points for validation.

Returns:

Dictionary mapping contract names to their entry points

Return type:

Dict[str, str]

validate_contract_script_mapping()

Validate contract-script relationships across the system.

Returns:

Dictionary with validation results and mapping statistics

Return type:

Dict[str, Any]

get_job_type_variants(base_step_name)

Get all job_type variants for a base step name.

Parameters:

base_step_name (str) – Base name of the step

Returns:

List of job type variants found

Return type:

List[str]

resolve_pipeline_node(node_name)

Resolve PipelineDAG node name to StepInfo (handles job_type variants).

Parameters:

node_name (str) – Node name from PipelineDAG

Returns:

StepInfo for the node, or None if not found

Return type:

StepInfo | None

get_builder_for_config(config, node_name=None)

Map a config instance to a builder PROVIDER (the thing the assembler calls with its 5-kwarg signature to get a StepBuilderBase).

Returns a BuilderProvider (FZ 31e1d3g1 Phase 1): TODAY this is the per-step builder CLASS (a class is already such a callable), so behavior is unchanged. The annotation is provider-based so the classless Design-B end-state can return a non-class factory here without breaking the contract — callers must invoke the result, never assume it is a class.

Parameters:

• config – Configuration instance (BasePipelineConfig)

• node_name (str) – Optional DAG node name for context

Returns:

A builder provider (currently a builder class) or None if not found.

Return type:

Callable[[…], Any] | None

get_builder_for_step_type(step_type)

Get the builder PROVIDER for a step type, with legacy alias support.

Returns a BuilderProvider (FZ 31e1d3g1 Phase 1) — currently the builder class; see get_builder_for_config for the dual-mode contract.

Parameters:

step_type (str) – Step type name (may be legacy alias)

Returns:

A builder provider (currently a builder class) or None if not found.

Return type:

Callable[[…], Any] | None

is_step_type_supported(step_type)

Check if step type is supported (including legacy aliases).

Parameters:

step_type (str) – Step type name

Returns:

True if supported, False otherwise

Return type:

bool

validate_builder_availability(step_types)

Validate that builders are available for step types.

Parameters:

step_types (List[str]) – List of step types to validate

Returns:

Dictionary mapping step types to availability status

Return type:

Dict[str, bool]

get_config_types_for_step_type(step_type)

Get possible config class names for a step type.

Parameters:

step_type (str) – Step type name

Returns:

List of possible configuration class names

Return type:

List[str]

list_supported_step_types()

List all supported step types including legacy aliases.

Returns:

List of supported step type names

Return type:

List[str]

validate_step_name_with_registry(step_name)

Use registry system for step name validation.

Parameters:

step_name (str) – Step name to validate

Returns:

True if valid, False otherwise

Return type:

bool

get_builder_map()

Get a complete builder map for pipeline construction.

Returns:

Dictionary mapping step types to builder classes

Return type:

Dict[str, Type]

validate_dag_compatibility(step_types)

Validate DAG compatibility with available builders.

Parameters:

step_types (List[str]) – List of step types in the DAG

Returns:

Dictionary with validation results

Return type:

Dict[str, Any]

get_step_builder_suggestions(config_class_name)

Get suggestions for step builders based on config class name.

Parameters:

config_class_name (str) – Configuration class name

Returns:

List of suggested step type names

Return type:

List[str]

get_metrics_report()

Get simple metrics report.

get_all_builders()

Get all available builders with canonical names.

This method provides comprehensive builder discovery for dynamic testing without requiring hard-coded maintenance.

Returns:

Dict mapping canonical names to builder classes

Return type:

Dict[str, Type]

get_builders_by_step_type(step_type)

Get builders filtered by SageMaker step type.

This method enables step-type-specific testing by filtering builders based on their registered SageMaker step type.

Parameters:

step_type (str) – SageMaker step type (Processing, Training, Transform, CreateModel, etc.)

Returns:

Dict mapping canonical names to builder classes for the step type

Return type:

Dict[str, Type]

get_immediate_parent_config_class(config_class_name)

Get the immediate parent config class name for inheritance.

This method enables Smart Default Value Inheritance by identifying which parent config class a specific config should inherit from. It implements cascading inheritance where TabularPreprocessingConfig inherits from ProcessingStepConfigBase (not BasePipelineConfig directly) to get all cascaded values.

Parameters:

config_class_name (str) – Target config class name (e.g., “TabularPreprocessingConfig”)

Returns:

Immediate parent class name (e.g., “ProcessingStepConfigBase”) or None if not found

Return type:

str | None

Example

parent = step_catalog.get_immediate_parent_config_class(“TabularPreprocessingConfig”) # Returns: “ProcessingStepConfigBase” (not “BasePipelineConfig”)

parent = step_catalog.get_immediate_parent_config_class(“CradleDataLoadConfig”) # Returns: “BasePipelineConfig” (direct inheritance)

extract_parent_values_for_inheritance(target_config_class_name, completed_configs)

Extract parent values for inheritance from completed configs.

This method enables Smart Default Value Inheritance by extracting field values from the immediate parent config for pre-populating child config forms. It implements cascading inheritance to eliminate redundant user input.

Parameters:

• target_config_class_name (str) – Target config class name

• completed_configs (Dict[str, Any]) – Dictionary of completed config instances by class name

Returns:

Dictionary of field values from immediate parent config

Return type:

Dict[str, Any]

Example

completed_configs = {

“BasePipelineConfig”: base_config_instance, “ProcessingStepConfigBase”: processing_config_instance

}

parent_values = step_catalog.extract_parent_values_for_inheritance(

“TabularPreprocessingConfig”, completed_configs

) # Returns: ALL field values from ProcessingStepConfigBase # (which includes cascaded values from BasePipelineConfig) # Result: {“author”: “lukexie”, “bucket”: “my-bucket”, # “processing_instance_type”: “ml.m5.2xlarge”, …}

discover_script_files(project_id=None)

Discover script files from package and workspaces with prioritization.

This method enables script discovery for interactive runtime testing by finding scripts referenced in config and contract entry points.

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping script names to ScriptInfo objects

Return type:

Dict[str, Any]

discover_scripts_from_dag(dag)

Discover scripts referenced in a DAG with intelligent node-to-script mapping.

This method enables DAG-guided script discovery for interactive runtime testing by mapping DAG nodes to actual script files using step catalog intelligence.

Parameters:

dag – PipelineDAG object

Returns:

Dictionary mapping script names to ScriptInfo objects

Return type:

Dict[str, Any]

load_script_info(script_name)

Load script information for a specific script with workspace-aware discovery.

This method enables script information retrieval for interactive runtime testing with workspace prioritization support.

Parameters:

script_name (str) – Name of the script to load info for

Returns:

ScriptInfo object or None if not found

Return type:

Any | None

get_script_info(script_name)

Get information about a script in dictionary format.

This method provides script information for interactive runtime testing in a user-friendly dictionary format.

Parameters:

script_name (str) – Name of the script

Returns:

Dictionary with script information or None if not found

Return type:

Dict[str, Any] | None

list_available_scripts()

List all available script names.

This method enables script enumeration for interactive runtime testing by listing all discovered script names.

Returns:

List of script names that have been discovered

Return type:

List[str]

get_script_discovery_stats()

Get script discovery statistics.

This method provides discovery statistics for interactive runtime testing to help users understand the script discovery process.

Returns:

Dictionary with script discovery statistics

Return type:

Dict[str, Any]