cursus.registry.hybrid

Phase 3: Simplified Local Registry Infrastructure

This module provides the core hybrid registry system components for Phase 3 implementation. Focuses on essential functionality with reduced redundancy.

Architecture: - utils.py: Simple function-based utilities - models.py: Pydantic data models - manager.py: Unified registry manager - setup.py: Workspace initialization utilities

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(*, step_name, resolved, selected_definition=None, reason=None, conflicting_definitions=<factory>, source_registry, resolution_strategy, workspace_id=None, conflict_detected=False, conflict_analysis=None, errors=<factory>, warnings=<factory>, resolution_metadata=<factory>, **extra_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 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.

CoreStepRegistry

alias of UnifiedRegistryManager

LocalStepRegistry

alias of UnifiedRegistryManager

HybridRegistryManager

alias of UnifiedRegistryManager

load_registry_module(file_path)

Load registry module from file.

Parameters:

file_path (str) – Path to the registry file

Returns:

Loaded module object

Raises:

RegistryLoadError – If module loading fails

Return type:

Any

from_legacy_format(step_name, step_info, registry_type='core', workspace_id=None)

Convert legacy STEP_NAMES format to StepDefinition.

Parameters:

• step_name (str) – Name of the step

• step_info (Dict[str, Any]) – Legacy step information dictionary

• registry_type (str) – Type of registry (‘core’, ‘workspace’, ‘override’)

• workspace_id (str) – Workspace identifier for workspace steps

Returns:

StepDefinition object

Return type:

StepDefinition

to_legacy_format(definition)

Convert StepDefinition to legacy STEP_NAMES format using field list.

Parameters:

definition (StepDefinition) – StepDefinition object

Returns:

Legacy format dictionary

Return type:

Dict[str, Any]

convert_registry_dict(registry_dict, registry_type='core', workspace_id=None)

Convert a complete registry dictionary to StepDefinition objects.

Parameters:

• registry_dict (Dict[str, Dict[str, Any]]) – Dictionary of step_name -> step_info

• registry_type (str) – Type of registry

• workspace_id (str) – Workspace identifier

Returns:

Dictionary of step_name -> StepDefinition

Return type:

Dict[str, StepDefinition]

validate_registry_data(registry_type, step_name, workspace_id=None)

Validate registry data using direct validation functions.

Parameters:

• registry_type (str) – Registry type to validate

• step_name (str) – Step name to validate

• workspace_id (str) – Optional workspace ID to validate

Returns:

True if valid

Raises:

ValueError – If validation fails

Return type:

bool

format_step_not_found_error(step_name, workspace_context=None, available_steps=None)

Format step not found error messages using generic formatter.

format_registry_load_error(registry_path, error_details)

Format registry loading error messages using generic formatter.

Modules

manager	Unified Registry Manager Implementation
models	Simplified Data Models for Hybrid Registry System
setup	Workspace registry initialization utilities.
utils	Streamlined Utility Functions for Hybrid Registry System