cursus.step_catalog

Unified Step Catalog System

This module provides a unified interface for discovering and retrieving step-related components (scripts, contracts, specifications, builders, configs) across multiple workspaces.

The system consolidates 16+ fragmented discovery mechanisms into a single, efficient StepCatalog class that provides O(1) lookups and intelligent component discovery.

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'}

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]

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]

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

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]

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

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]

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”, …}

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]

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]

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

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_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

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

get_builder_map()

Get a complete builder map for pipeline construction.

Returns:

Dictionary mapping step types 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_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]

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]

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)

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]

get_metrics_report()

Get simple metrics report.

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]

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

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

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_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

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

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

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

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]

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

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

List all supported step types including legacy aliases.

Returns:

List of supported step type names

Return type:

List[str]

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

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

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_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

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

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

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]

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]

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]

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]

validate_contract_script_mapping()

Validate contract-script relationships across the system.

Returns:

Dictionary with validation results and mapping statistics

Return type:

Dict[str, Any]

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]

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

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

class StepInfo(*, step_name, workspace_id, registry_data=<factory>, file_components=<factory>)

Bases: BaseModel

Essential step information combining registry data with file metadata.

property builder_step_name: str

Get builder step name from registry data.

property config_class: str

Get config class name from registry data.

property description: str

Get step description from registry data.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

property sagemaker_step_type: str

Get SageMaker step type from registry data.

step_name: str

workspace_id: str

registry_data: Dict[str, Any]

file_components: Dict[str, FileMetadata | None]

class FileMetadata(*, path, file_type, modified_time)

Bases: BaseModel

Simple metadata for component files.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

path: Path

file_type: str

modified_time: datetime

class StepSearchResult(*, step_name, workspace_id, match_score, match_reason, components_available=<factory>)

Bases: BaseModel

Simple search result for step queries.

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

step_name: str

workspace_id: str

match_score: float

match_reason: str

components_available: List[str]

class ConfigAutoDiscovery(package_root, workspace_dirs)

Bases: object

Simple configuration class auto-discovery.

build_complete_config_classes(project_id=None)

Build complete mapping using pure auto-discovery. Includes both config and hyperparameter classes for comprehensive discovery.

ConfigClassStore was removed as part of the unified step catalog refactoring. This method now uses pure AST-based auto-discovery which is deployment-agnostic and works in all environments (installed, submodule, Lambda, Docker, etc.).

Parameters:

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

Returns:

Complete dictionary of config and hyperparameter classes (auto-discovered)

Return type:

Dict[str, Type]

discover_config_classes(project_id=None)

Auto-discover configuration classes from package 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]

discover_hyperparameter_classes(project_id=None)

Auto-discover hyperparameter classes from package 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]

create_step_catalog(workspace_root, use_unified=None)

Factory function for step catalog with feature flag support.

Parameters:

• workspace_root (Path) – Root directory of the workspace

• use_unified (bool | None) – Whether to use unified catalog (None = check environment)

Returns:

StepCatalog instance or legacy wrapper based on feature flag

Return type:

Any

class ContractDiscoveryEngineAdapter(workspace_root)

Bases: object

Modernized adapter using unified StepCatalog for all discovery operations.

Replaces: src/cursus/validation/alignment/discovery/contract_discovery.py All methods now use the step catalog’s built-in discovery capabilities.

build_entry_point_mapping()

MODERNIZED: Build entry point mapping from step names to script paths.

Used by tests and legacy systems that expect entry point mappings.

Returns:

Dictionary mapping step names to script file paths

Return type:

Dict[str, str]

discover_all_contracts()

MODERNIZED: Discover all steps that have contracts using step catalog.

Used by alignment validation system.

discover_contracts_with_scripts()

MODERNIZED: Use step catalog’s built-in method.

Historically the primary contract-discovery entry for the Level-2 alignment tester (now removed — contract<->spec alignment is a construction-time invariant; FZ 31e1d3h/D5); the method stays as a general script-backed-contract discovery helper.

extract_contract_reference_from_spec(spec_file)

MODERNIZED: Extract contract reference from spec file using step catalog.

Used by alignment validation system to find which step corresponds to a spec file.

Parameters:

spec_file (str) – Name of the spec file (e.g., “xgboost_training_spec.py”)

Returns:

Step name if found, None otherwise

Return type:

str | None

class ContractDiscoveryManagerAdapter(test_data_dir=None, workspace_root=None)

Bases: object

Modernized adapter using unified StepCatalog with minimal business logic.

Replaces: src/cursus/validation/script_testing (formerly runtime/contract_discovery.py) Focuses on test-specific functionality while leveraging step catalog for discovery.

discover_contract(step_name, canonical_name=None)

MODERNIZED: Use StepCatalog’s load_contract_class for discovery.

Parameters:

• step_name (str) – Name of the step/script

• canonical_name (str | None) – Optional canonical name for the step

Returns:

String path for backward compatibility, or None if not found

Return type:

str | None

get_contract_environ_vars(contract)

Get environment variables from contract.

get_contract_input_paths(contract, step_name)

Get contract input paths with local adaptation.

get_contract_job_args(contract, step_name)

Get job arguments from contract.

get_contract_output_paths(contract, step_name)

Get contract output paths with local adaptation.

class ContractDiscoveryResult(contract=None, contract_name=None, discovery_method='step_catalog', error_message=None)

Bases: object

Legacy result class for contract discovery operations.

Maintains backward compatibility with existing tests and code.

class FlexibleFileResolverAdapter(workspace_root)

Bases: object

Modernized file resolver using unified step catalog system.

Replaces: src/cursus/validation/alignment/file_resolver.py

This adapter leverages the step catalog’s superior discovery capabilities while maintaining backward compatibility with legacy FlexibleFileResolver APIs.

extract_base_name_from_spec(spec_path)

Extract base name from spec path.

find_all_component_files(step_name)

Legacy method: find all component files for step.

find_builder_file(step_name)

Legacy method: find builder file for step.

find_config_file(step_name)

Legacy method: find config file for step.

find_contract_file(step_name)

Legacy method: find contract file for step.

find_spec_constant_name(script_name, job_type='training')

Find spec constant name using catalog.

find_spec_file(step_name)

Legacy method: find spec file for step.

find_specification_file(script_name)

Legacy alias for find_spec_file.

get_available_files_report()

Generate report using catalog data.

refresh_cache()

Legacy method: refresh cache using catalog.

class DeveloperWorkspaceFileResolverAdapter(workspace_root=None, project_id=None, developer_id=None, enable_shared_fallback=True, **kwargs)

Bases: FlexibleFileResolverAdapter

Adapter maintaining backward compatibility with DeveloperWorkspaceFileResolver.

Replaces: src/cursus/workspace/validation/workspace_file_resolver.py

find_builder_file(step_name)

Workspace-aware builder file discovery. Returns string path like legacy method.

find_config_file(step_name)

Workspace-aware config file discovery. Returns string path like legacy method.

find_contract_file(step_name)

Workspace-aware contract file discovery. Returns string path like legacy method.

find_script_file(step_name, script_name=None)

Workspace-aware script file discovery. Returns string path like legacy method.

find_spec_file(step_name)

Workspace-aware spec file discovery. Returns string path like legacy method.

get_workspace_info()

Get workspace information using step catalog.

list_available_developers()

List available developers using workspace discovery.

switch_developer(developer_id)

Switch to a different developer workspace.

class WorkspaceDiscoveryManagerAdapter(workspace_root)

Bases: object

Adapter maintaining backward compatibility with WorkspaceDiscoveryManager.

Replaces: src/cursus/workspace/core/discovery.py

discover_components(workspace_ids=None, developer_id=None)

Enhanced method: discover ALL component types with workspace-specific functionality.

When workspace_ids is None: focuses on shared package (cursus/steps - “core” workspace) When workspace_ids specified: discovers from those specific workspaces

Returns comprehensive inventory of all 5 component types: - builders, configs, contracts, specs, scripts

discover_workspaces(workspace_root)

Legacy method: discover available workspaces with simplified structure.

get_discovery_summary()

Legacy method: get discovery summary.

get_file_resolver(developer_id=None, **kwargs)

Legacy method: get file resolver.

get_module_loader(developer_id=None, **kwargs)

Legacy method: get module loader.

get_statistics()

Legacy method: get statistics.

get_workspace_info(workspace_id=None, developer_id=None)

Legacy method: get workspace info with simplified structure.

list_available_developers()

Legacy method: list available developers with simplified structure.

refresh_cache()

Legacy method: refresh cache.

class HybridFileResolverAdapter(workspace_root)

Bases: object

Adapter maintaining backward compatibility with HybridFileResolver.

Replaces: src/cursus/validation/alignment/patterns/file_resolver.py

resolve_file_pattern(pattern, component_type)

Legacy method: resolve files matching pattern.

class LegacyDiscoveryWrapper(workspace_root)

Bases: object

Wrapper providing legacy discovery interfaces during migration period.

This class provides a unified interface that can be used as a drop-in replacement for legacy discovery systems during the migration phase. It delegates all StepCatalog methods to the underlying catalog while also providing access to legacy adapters.

build_complete_config_classes(project_id=None)

Delegate to underlying StepCatalog.

detect_framework(step_name)

Delegate to underlying StepCatalog.

discover_config_classes(project_id=None)

Delegate to underlying StepCatalog.

discover_contracts_with_scripts()

Delegate to underlying StepCatalog.

discover_cross_workspace_components(workspace_ids=None)

Delegate to underlying StepCatalog.

extract_base_name_from_spec(spec_path)

Extract base name from spec path.

find_all_component_files(step_name)

Legacy method: find all component files for step.

find_builder_file(step_name)

Legacy method: find builder file for step.

find_config_file(step_name)

Legacy method: find config file for step.

find_contract_file(step_name)

Legacy method: find contract file for step.

find_spec_constant_name(script_name, job_type='training')

Find spec constant name using catalog.

find_spec_file(step_name)

Legacy method: find spec file for step.

find_specification_file(script_name)

Legacy alias for find_spec_file.

find_step_by_component(component_path)

Delegate to underlying StepCatalog.

get_adapter(adapter_type)

Get specific legacy adapter by type.

get_available_files_report()

Generate report using catalog data.

get_builder_class_path(step_name)

Delegate to underlying StepCatalog.

get_job_type_variants(step_name)

Delegate to underlying StepCatalog.

get_metrics_report()

Delegate to underlying StepCatalog.

get_step_info(step_name, job_type=None)

Delegate to underlying StepCatalog.

get_unified_catalog()

Get the underlying unified catalog.

list_available_steps(workspace_id=None, job_type=None)

Delegate to underlying StepCatalog.

load_builder_class(step_name)

Delegate to underlying StepCatalog.

refresh_cache()

Legacy method: refresh cache using catalog.

search_steps(query, job_type=None)

Delegate to underlying StepCatalog.

Modules

adapters	Step catalog adapters package.
builder_discovery	Builder class discovery via registry-interface synthesis.
config_discovery	Configuration class auto-discovery for the unified step catalog system.
contract_discovery	Contract discovery for the unified step catalog system.
mapping	Step Catalog Mapping Module - Config-to-Builder Resolution and Legacy Support.
models	Simple data models for the unified step catalog system.
naming	Single source of truth for step-name <-> file-name conversion in the step catalog.
script_discovery	Script file auto-discovery for the unified step catalog system.
sdk_bindings	Lazy SAIS-SDK step-class bindings for the registry-walk builder materializer (FZ 31e1d3g3 Phase A2).
spec_discovery	Specification discovery for the unified step catalog system.
step_catalog	Unified Step Catalog - Single class addressing all US1-US5 requirements.