cursus.registry.hybrid.models

Simplified Data Models for Hybrid Registry System

This module contains essential Pydantic data models for the Phase 3 simplified hybrid registry system with optimized validation using enums.

Models: - StepDefinition: Core step definition with registry metadata - ResolutionContext: Context for step resolution - StepResolutionResult: Result of step resolution - RegistryValidationResult: Results of registry validation - ConflictAnalysis: Basic analysis of step name conflicts

class RegistryType(*values)[source]

Bases: str, Enum

Registry type enumeration for automatic validation.

CORE = 'core'
WORKSPACE = 'workspace'
OVERRIDE = 'override'
class ResolutionMode(*values)[source]

Bases: str, Enum

Resolution mode enumeration for automatic validation.

AUTOMATIC = 'automatic'
INTERACTIVE = 'interactive'
STRICT = 'strict'
class ResolutionStrategy(*values)[source]

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)[source]

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'
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>)[source]

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

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]
classmethod validate_identifiers(v)[source]

Validate identifier fields using simple validation.

property step_name: str

Alias for name attribute to maintain compatibility with step catalog.

to_legacy_format()[source]

Convert to legacy STEP_NAMES format using shared converter.

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

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)[source]

Bases: BaseModel

Result of step conflict resolution using Pydantic V2.

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]
get_resolution_summary()[source]

Get a summary of the resolution result.

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

Bases: BaseModel

Results of registry validation using Pydantic V2.

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
get_validation_summary()[source]

Get a summary of validation results.

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)[source]

Bases: BaseModel

Analysis of a step name conflict using Pydantic V2.

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
get_conflict_summary()[source]

Get a summary of the conflict.