cursus.registry

Enhanced Pipeline Registry Module with Hybrid Registry Support.

This module contains registry components for tracking step types, specifications, hyperparameters, and other metadata used in the pipeline system. It helps ensure consistency in step naming and configuration.

Enhanced Features: - Workspace-aware step resolution - Hybrid registry backend support - Context management for multi-developer workflows - Backward compatibility with existing code

exception RegistryError(message, unresolvable_types=None, available_builders=None)

Bases: Exception

Raised when step builder registry errors occur.

class KnobSpec(name, type='str', default=None, required=False, doc='')

Bases: object

Describes one tunable knob a handler accepts.

default: Any = None

doc: str = ''

required: bool = False

type: str = 'str'

name: str

class StrategyInfo(axis, name, handler, knobs=(), preset_knobs=<factory>, routable=True, verb='', implemented=True)

Bases: object

A registry row: a routing (axis, name) -> a handler class + its knobs.

implemented: bool = True

knobs: Tuple[KnobSpec, ...] = ()

routable: bool = True

verb: str = ''

axis: str

name: str

handler: Callable[..., 'PatternHandler'] | None

preset_knobs: Dict[str, Any]

exception NoBuilderError

Bases: ValueError

Raised when an (axis, name) has no routable strategy (abstract / builder-less type).

register_strategy(*, axis, name, knobs=(), preset_knobs=None, routable=True, verb='', implemented=True)

Decorator: register a handler class under (axis, name). Returns the class unchanged.

register_no_builder(*, axis, name, verb='')

Register a non-routable row (e.g. Base / Lambda — abstract or builder-less types).

resolve_strategy(axis, name)

Return the routable StrategyInfo for (axis, name), else raise NoBuilderError.

list_strategies(axis=None)

All registered strategies, optionally filtered by axis.

knobs_for(axis, name)

The declarative knobs the strategy at (axis, name) accepts.

axes()

The routing axes that have registered strategies.

axis_name_for_step_type(sagemaker_step_type, step_assembly=None)

Map a step’s (sagemaker_step_type, step_assembly) onto the registry (axis, name) key.

This is the single source of the routing rule — builder_templates.resolve_handler calls it to bind the handler at build time, and the introspection tool calls it for for_step_type — so the tool’s “what would this step bind?” answer can never diverge from the actual router.

Routing is by sagemaker_step_type ONLY (never by step name); Processing is the one type sub-discriminated by step_assembly (code | step_args | delegation, default code).

find_strategies(name, axis=None)

All registered strategies whose name matches, optionally constrained to one axis.

Names are unique within an axis but a bare name may (in principle) appear on more than one axis, so callers that need exactly one row should disambiguate with axis.

strategy_to_dict(info)

A StrategyInfo as a plain, JSON-serializable dict (handler rendered as its class name).

knob_to_dict(knob)

A KnobSpec as a plain, JSON-serializable dict.

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.

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.

set_workspace_context(workspace_id)

Set workspace context and refresh module variables.

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

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.

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.

class UnifiedRegistryManager(core_registry_path=None, workspaces_root=None)

Bases: object

Unified registry manager that consolidates all registry operations.

Replaces CoreStepRegistry, LocalStepRegistry, and HybridRegistryManager with a single, efficient manager that eliminates redundancy while maintaining all functionality.

add_workspace_registry(workspace_id, workspace_path)

Add a new workspace registry.

clear_builder_class_cache()

Clear builder class cache.

clear_component_cache()

Clear component discovery cache.

clear_workspace_context()

Clear current workspace context.

create_legacy_step_names_dict(workspace_id=None)

Create legacy STEP_NAMES dictionary for backward compatibility with caching.

get_all_step_definitions(workspace_id=None)

Get all step definitions with caching for performance optimization.

get_builder_class_cache(cache_key)

Get cached builder class.

get_component_cache(cache_key)

Get cached component discovery results.

get_local_only_definitions(workspace_id)

Get only local and override definitions for a workspace (not core).

get_registry_status()

Get status information for all registries.

get_step(step_name, context=None)

Get a step definition with simple workspace priority resolution.

Parameters:

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

• context (ResolutionContext | None) – Resolution context for workspace handling

Returns:

StepResolutionResult containing the resolved step and metadata

Return type:

StepResolutionResult

get_step_conflicts()

Identify steps defined in multiple registries.

get_step_count(workspace_id=None)

Get the total number of steps in the registry.

get_step_definition(step_name, workspace_id=None)

Get a step definition by name, with optional workspace context.

Parameters:

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

• workspace_id (str) – Optional workspace context for resolution

Returns:

StepDefinition if found, None otherwise

Return type:

StepDefinition | None

get_workspace_context()

Get current workspace context.

has_step(step_name, workspace_id=None)

Check if a step exists in the registry.

list_all_steps(include_source=False)

List all available steps across all registries.

list_steps(workspace_id=None)

List all available step names for a workspace or core.

reload_core_registry()

Re-snapshot the core registry from step_names_base.STEP_NAMES and drop caches.

Called after an external step-pack merges rows into step_names_base (see cursus.registry.step_names.refresh_registry()) so the manager’s _core_steps snapshot and its legacy/definition caches reflect the plugin steps. Additive by construction — it re-reads the (already package-first-merged) base table.

remove_workspace_registry(workspace_id)

Remove a workspace registry.

resolution_context(workspace_id)

Context manager for step resolution.

set_builder_class_cache(cache_key, builder_class)

Cache builder class.

set_component_cache(cache_key, components)

Cache component discovery results.

set_workspace_context(workspace_id)

Set current workspace context.

class StepDefinition(*, name, registry_type, config_class=None, spec_type=None, sagemaker_step_type=None, builder_step_name=None, description=None, framework=None, job_types=<factory>, workspace_id=None, override_source=None, priority=100, compatibility_tags=<factory>, framework_version=None, environment_tags=<factory>, conflict_resolution_strategy=ResolutionStrategy.WORKSPACE_PRIORITY, metadata=<factory>)

Bases: BaseModel

Enhanced step definition with registry metadata using Pydantic V2 and enum validation.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': False, 'str_strip_whitespace': True, 'validate_assignment': True}

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

property step_name: str

Alias for name attribute to maintain compatibility with step catalog.

to_legacy_format()

Convert to legacy STEP_NAMES format using shared converter.

classmethod validate_identifiers(v)

Validate identifier fields using simple validation.

name: str

registry_type: RegistryType

config_class: str | None

spec_type: str | None

sagemaker_step_type: str | None

builder_step_name: str | None

description: str | None

framework: str | None

job_types: List[str]

workspace_id: str | None

override_source: str | None

priority: int

compatibility_tags: List[str]

framework_version: str | None

environment_tags: List[str]

conflict_resolution_strategy: ResolutionStrategy

metadata: Dict[str, Any]

class ResolutionContext(*, workspace_id=None, preferred_framework=None, environment_tags=<factory>, resolution_mode=ResolutionMode.AUTOMATIC, resolution_strategy=ResolutionStrategy.WORKSPACE_PRIORITY)

Bases: BaseModel

Context for step resolution and conflict resolution using Pydantic V2 and enum validation.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'frozen': False, 'validate_assignment': True}

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

workspace_id: str | None

preferred_framework: str | None

environment_tags: List[str]

resolution_mode: ResolutionMode

resolution_strategy: ResolutionStrategy

class StepResolutionResult(**data)

Bases: BaseModel

Result of step conflict resolution using Pydantic V2.

get_resolution_summary()

Get a summary of the resolution result.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'frozen': False, 'validate_assignment': True}

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

step_name: str

resolved: bool

selected_definition: StepDefinition | None

reason: str | None

conflicting_definitions: List[StepDefinition]

source_registry: str

resolution_strategy: str

workspace_id: str | None

conflict_detected: bool

conflict_analysis: ConflictAnalysis | None

errors: List[str]

warnings: List[str]

resolution_metadata: Dict[str, Any]

class RegistryValidationResult(*, is_valid, registry_type='unknown', issues=<factory>, errors=<factory>, warnings=<factory>, step_count=0, **extra_data)

Bases: BaseModel

Results of registry validation using Pydantic V2.

get_validation_summary()

Get a summary of validation results.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'frozen': False, 'validate_assignment': True}

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

is_valid: bool

registry_type: str

issues: List[str]

errors: List[str]

warnings: List[str]

step_count: int

class ConflictAnalysis(*, step_name, conflicting_definitions=<factory>, resolution_strategies=<factory>, recommended_strategy=None, impact_assessment=None, conflicting_sources, resolution_strategy, workspace_context=None, conflict_type='name_conflict', **extra_data)

Bases: BaseModel

Analysis of a step name conflict using Pydantic V2.

get_conflict_summary()

Get a summary of the conflict.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'frozen': False, 'validate_assignment': True}

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

step_name: str

conflicting_definitions: List[StepDefinition]

resolution_strategies: List[str]

recommended_strategy: str | None

impact_assessment: str | None

conflicting_sources: List[str]

resolution_strategy: str

workspace_context: str | None

conflict_type: str

class RegistryType(*values)

Bases: str, Enum

Registry type enumeration for automatic validation.

CORE = 'core'

WORKSPACE = 'workspace'

OVERRIDE = 'override'

class ResolutionMode(*values)

Bases: str, Enum

Resolution mode enumeration for automatic validation.

AUTOMATIC = 'automatic'

INTERACTIVE = 'interactive'

STRICT = 'strict'

class ResolutionStrategy(*values)

Bases: str, Enum

Resolution strategy enumeration for automatic validation.

WORKSPACE_PRIORITY = 'workspace_priority'

FRAMEWORK_MATCH = 'framework_match'

ENVIRONMENT_MATCH = 'environment_match'

MANUAL = 'manual'

HIGHEST_PRIORITY = 'highest_priority'

CORE_FALLBACK = 'core_fallback'

class ConflictType(*values)

Bases: str, Enum

Conflict type enumeration for automatic validation.

NAME_CONFLICT = 'name_conflict'

FRAMEWORK_CONFLICT = 'framework_conflict'

VERSION_CONFLICT = 'version_conflict'

WORKSPACE_CONFLICT = 'workspace_conflict'

switch_to_workspace(workspace_id)

Switch to a specific workspace context.

Parameters:

workspace_id (str) – Workspace identifier to switch to

Example

switch_to_workspace(“developer_1”) step_names = STEP_NAMES # Now uses developer_1 context

switch_to_core()

Switch back to core registry (no workspace context).

Example

switch_to_core() step_names = STEP_NAMES # Now uses core registry only

get_registry_info(workspace_id=None)

Get comprehensive registry information for a workspace or core.

Parameters:

workspace_id (str) – Optional workspace identifier

Returns:

Dictionary with registry information

Return type:

dict

Modules

exceptions	Exception classes for the Pipeline Registry.
hybrid	Phase 3: Simplified Local Registry Infrastructure
interface_registry_loader	Interface-derived step registry loader — the SOLE source of the step-name registry.
step_names	Enhanced step names registry with hybrid backend support.
step_names_base	Base step-name registry — the single source of truth for step naming across config, builders, and specifications.
strategy_registry	Strategy registry — the single source of truth for the builder strategy library.
validation_utils	Simple validation utilities for step definition standardization.