cursus.registry.step_names

Enhanced step names registry with hybrid backend support. Maintains 100% backward compatibility while adding workspace awareness.

This module provides a drop-in replacement for the original step_names.py that: - Uses the hybrid registry backend transparently - Maintains all existing functions and variables - Adds workspace context management capabilities - Provides seamless workspace-aware step resolution

get_workspace_context()

Get current workspace context.

Returns:

Current workspace identifier or None if no context set

Return type:

str | None

clear_workspace_context()

Clear current workspace context.

workspace_context(workspace_id)

Context manager for temporary workspace context.

Parameters:

workspace_id (str) – Workspace identifier for temporary context

Usage:

with workspace_context(“developer_1”):

step_names = get_step_names() # Uses developer_1 context

is_hybrid_active()

Return True if the hybrid (workspace-aware) registry manager is in use.

False means initialization failed and the static fallback registry is active — see get_registry_health() for the captured error.

get_registry_health()

Report registry-manager health so callers can detect a silent fallback.

Returns a dict with hybrid_active and, when degraded, init_error (the stringified exception that forced the static fallback).

refresh_registry(pack_interfaces_dir=None)

Merge an external step-pack’s .step.yaml interfaces INTO the step registry (add-only).

This is the public entry point behind action item AI-2. It enforces the additive invariant: package steps are ALWAYS available; a pack can only ADD steps (and, on a deliberate name-clash, shadow with a warning). It never removes or replaces a package step.

Mechanism (package-first, never replace):

1. Derive the pack’s registry rows from pack_interfaces_dir/*.step.yaml via build_registry_from_interfaces(pack_interfaces_dir).

2. Layer them on top of the live package table with step_names_base.merge_pack_registry (in-place STEP_NAMES.update — package rows preserved).

3. Re-sync the hybrid manager (reload_core_registry) so the StepCatalog, which reads get_step_names() through the manager, sees the plugin steps.

Parameters:

pack_interfaces_dir (Any | None) – directory of the pack’s *.step.yaml files (e.g. <project_root>/step_pack/interfaces). None is a no-op returning {}.

Returns:

{name: "collision"} for pack names that shadowed an existing package step (already logged as warnings here); empty when every pack step is new.

Return type:

Dict[str, str]

get_step_names(workspace_id=None)

Get STEP_NAMES dictionary with workspace context.

Parameters:

workspace_id (str) – Optional workspace context override

Returns:

Step names dictionary in original format

Return type:

Dict[str, Dict[str, str]]

get_config_step_registry(workspace_id=None)

Get CONFIG_STEP_REGISTRY with workspace context.

get_builder_step_names(workspace_id=None)

Get BUILDER_STEP_NAMES with workspace context.

get_spec_step_types(workspace_id=None)

Get SPEC_STEP_TYPES with workspace context.

get_config_class_name(step_name, workspace_id=None)

Get config class name with workspace context.

get_builder_step_name(step_name, workspace_id=None)

Get builder step class name with workspace context.

get_spec_step_type(step_name, workspace_id=None)

Get step_type value for StepSpecification with workspace context.

get_spec_step_type_with_job_type(step_name, job_type=None, workspace_id=None)

Get step_type with optional job_type suffix, workspace-aware.

get_step_name_from_spec_type(spec_type, workspace_id=None)

Get canonical step name from spec_type with workspace context.

get_all_step_names(workspace_id=None)

Get all canonical step names with workspace context.

validate_step_name(step_name, workspace_id=None)

Validate step name exists with workspace context.

validate_spec_type(spec_type, workspace_id=None)

Validate spec_type exists with workspace context.

get_step_description(step_name, workspace_id=None)

Get step description with workspace context.

list_all_step_info(workspace_id=None)

Get complete step information with workspace context.

get_sagemaker_step_type(step_name, workspace_id=None)

Get SageMaker step type with workspace context.

get_steps_by_sagemaker_type(sagemaker_type, workspace_id=None)

Get steps by SageMaker type with workspace context.

get_all_sagemaker_step_types(workspace_id=None)

Get all SageMaker step types with workspace context.

get_valid_sagemaker_step_types(workspace_id=None)

The authoritative set of valid SageMaker step types — single source of truth (FZ 31e1d3g3 C3).

Union of the LIVE types declared across every step’s .step.yaml (robust to shell deletion, since it reads the interface-derived registry) and a small static floor of framework-structural types that may not be attached to a live step. Replaces the two divergent hardcoded sets.

validate_sagemaker_step_type(sagemaker_type, workspace_id=None)

Validate SageMaker step type against the single-source valid set.

get_sagemaker_step_type_mapping(workspace_id=None)

Get SageMaker step type mapping with workspace context.

get_canonical_name_from_file_name(file_name, workspace_id=None)

Enhanced file name resolution with workspace context awareness.

Parameters:

• file_name (str) – File-based name (e.g., “model_evaluation_xgb”, “dummy_training”)

• workspace_id (str) – Optional workspace context for resolution

Returns:

Canonical step name (e.g., “XGBoostModelEval”, “DummyTraining”)

Raises:

ValueError – If file name cannot be mapped to a canonical name

Return type:

str

validate_file_name(file_name, workspace_id=None)

Validate file name can be mapped with workspace context.

list_available_workspaces()

List all available workspace contexts.

get_workspace_step_count(workspace_id)

Get number of steps available in a workspace.

has_workspace_conflicts()

Check if there are any step name conflicts between workspaces.

original_set_workspace_context(workspace_id)

Set current workspace context for registry resolution.

Parameters:

workspace_id (str) – Workspace identifier to set as current context

set_workspace_context(workspace_id)

Set workspace context and refresh module variables.

add_new_step_with_validation(step_name, config_class, builder_name, sagemaker_type, description='', validation_mode='warn', workspace_id=None)

Add new step with standardization validation.

This function implements the essential validation for new step creation identified in the redundancy analysis, providing a simple interface for adding steps with automatic validation.

Parameters:

• step_name (str) – Canonical step name (should be PascalCase)

• config_class (str) – Configuration class name (should end with ‘Config’)

• builder_name (str) – Builder class name (should end with ‘StepBuilder’)

• sagemaker_type (str) – SageMaker step type

• description (str) – Optional step description

• validation_mode (str) – Validation mode (“warn”, “strict”, “auto_correct”)

• workspace_id (str) – Optional workspace context

Returns:

List of validation warnings/messages

Raises:

ValueError – If validation fails in strict mode

Return type:

List[str]

Example

warnings = add_new_step_with_validation(

“MyCustomStep”, “MyCustomStepConfig”, “MyCustomStepStepBuilder”, “Processing”, “Custom processing step”

)

validate_step_definition_data(step_name, step_data)

Validate step definition data for standardization compliance.

This function provides the core validation functionality identified as essential in the redundancy analysis.

Parameters:

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

• step_data (Dict[str, str]) – Step definition data dictionary

Returns:

List of validation errors (empty if validation passes)

Return type:

List[str]

get_step_validation_suggestions(step_name, step_data)

Get detailed validation errors with helpful suggestions.

This function provides the clear error messages with examples identified as essential for developer experience.

Parameters:

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

• step_data (Dict[str, str]) – Step definition data dictionary

Returns:

List of detailed error messages with suggestions

Return type:

List[str]

auto_correct_step_data(step_name, step_data)

Auto-correct step definition data for common naming violations.

This function provides the simple auto-correction functionality identified as essential in the redundancy analysis.

Parameters:

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

• step_data (Dict[str, str]) – Step definition data dictionary

Returns:

Corrected step data dictionary

Return type:

Dict[str, str]

check_step_name_compliance(step_name, workspace_id=None)

Check if a step name complies with standardization rules.

Parameters:

• step_name (str) – Step name to check

• workspace_id (str) – Optional workspace context

Returns:

True if compliant, False otherwise

Return type:

bool

get_validation_status()

Get current validation system status.

Returns:

Dictionary with validation system information

Return type:

Dict[str, Any]