cursus.core.deps

Pipeline Dependencies module - Declarative dependency management for SageMaker pipelines.

This module provides declarative specifications and intelligent resolution for pipeline step dependencies.

class DependencyType(*values)

Bases: Enum

Types of dependencies in the pipeline.

MODEL_ARTIFACTS = 'model_artifacts'

PROCESSING_OUTPUT = 'processing_output'

TRAINING_DATA = 'training_data'

HYPERPARAMETERS = 'hyperparameters'

PAYLOAD_SAMPLES = 'payload_samples'

CUSTOM_PROPERTY = 'custom_property'

class NodeType(*values)

Bases: Enum

Types of nodes in the pipeline based on their dependency/output characteristics.

SOURCE = 'source'

INTERNAL = 'internal'

SINK = 'sink'

SINGULAR = 'singular'

class PropertyReference(*, step_name, output_spec)

Bases: BaseModel

Lazy evaluation reference bridging definition-time and runtime.

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

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

to_runtime_property(step_instances)

Create an actual SageMaker property reference using step instances.

This method navigates the property path to create a proper SageMaker Properties object that can be used at runtime.

Parameters:

step_instances (Dict[str, Any]) – Dictionary mapping step names to step instances

Returns:

SageMaker Properties object for the referenced property

Raises:

• ValueError – If the step is not found or property path is invalid

• AttributeError – If any part of the property path is invalid

Return type:

Any

to_sagemaker_property()

Convert to SageMaker Properties dictionary format at pipeline definition time.

classmethod validate_step_name(v)

Validate step name is not empty.

step_name: str

output_spec: OutputDecl

class SpecificationRegistry(context_name='default')

Bases: object

Context-aware registry for managing step specifications with isolation.

find_compatible_outputs(dependency_spec)

Find outputs compatible with a dependency specification.

get_specification(step_name)

Get specification by step name.

get_specifications_by_type(step_type)

Get all specifications of a given step type.

list_step_names()

Get list of all registered step names.

list_step_types()

Get list of all registered step types.

register(step_name, specification)

Register a step specification.

class RegistryManager(workspace_context=None)

Bases: object

Manager for context-scoped registries with complete isolation and workspace awareness.

This enhanced registry manager now supports: 1. Workspace-aware registry resolution 2. Integration with hybrid registry system 3. Backward compatibility with existing code 4. Thread-local workspace context management

clear_all_contexts()

Clear all registries.

clear_context(context_name)

Clear the registry for a specific context.

Parameters:

context_name (str) – Name of the context to clear

Returns:

True if the registry was cleared, False if it didn’t exist

Return type:

bool

get_context_stats()

Get statistics for all contexts.

Returns:

Dictionary mapping context names to their statistics

Return type:

Dict[str, Dict[str, int]]

get_registry(context_name='default', create_if_missing=True)

Get the registry for a specific context with workspace awareness.

This enhanced method now supports: 1. Workspace-aware context naming 2. Integration with hybrid registry system 3. Automatic registry population from hybrid sources 4. Backward compatibility with existing code

Parameters:

• context_name (str) – Name of the context (e.g., pipeline name, environment)

• create_if_missing (bool) – Whether to create a new registry if one doesn’t exist

Returns:

Context-specific registry or None if not found and create_if_missing is False

Return type:

SpecificationRegistry

list_contexts()

Get list of all registered context names.

Returns:

List of context names with registries

Return type:

List[str]

get_registry(manager=None, context_name='default')

Get the registry for a specific context.

Parameters:

• manager (RegistryManager | None) – Registry manager instance

• context_name (str) – Name of the context (e.g., pipeline name, environment)

Returns:

Context-specific registry

Return type:

SpecificationRegistry

get_pipeline_registry(manager, pipeline_name)

Get registry for a pipeline (backward compatibility).

Parameters:

• manager (RegistryManager) – Registry manager instance

• pipeline_name (str) – Name of the pipeline

Returns:

Pipeline-specific registry

Return type:

SpecificationRegistry

get_default_registry(manager)

Get the default registry (backward compatibility).

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

Default registry

Return type:

SpecificationRegistry

integrate_with_pipeline_builder(pipeline_builder_cls, manager=None)

Decorator to integrate context-scoped registries with a pipeline builder class.

This decorator modifies a pipeline builder class to use context-scoped registries.

Parameters:

• pipeline_builder_cls (Any) – Pipeline builder class to modify

• manager (RegistryManager | None) – Registry manager instance (if None, a new instance will be created)

Returns:

Modified pipeline builder class

Return type:

Any

list_contexts(manager)

Get list of all registered context names.

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

List of context names with registries

Return type:

List[str]

clear_context(manager, context_name)

Clear the registry for a specific context.

Parameters:

• manager (RegistryManager) – Registry manager instance

• context_name (str) – Name of the context to clear

Returns:

True if the registry was cleared, False if it didn’t exist

Return type:

bool

get_context_stats(manager)

Get statistics for all contexts.

Parameters:

manager (RegistryManager) – Registry manager instance

Returns:

Dictionary mapping context names to their statistics

Return type:

Dict[str, Dict[str, int]]

class UnifiedDependencyResolver(registry, semantic_matcher)

Bases: object

Intelligent dependency resolver using declarative specifications.

clear_cache()

Clear the resolution cache.

get_resolution_report(available_steps)

Generate a detailed resolution report for debugging.

Parameters:

available_steps (List[str]) – List of available step names

Returns:

Detailed report of resolution process

Return type:

Dict[str, Any]

register_specification(step_name, spec)

Register a step specification with the resolver.

resolve_all_dependencies(available_steps)

Resolve dependencies for all registered steps.

Parameters:

available_steps (List[str]) – List of step names that are available in the pipeline

Returns:

Dictionary mapping step names to their resolved dependencies

Return type:

Dict[str, Dict[str, PropertyReference]]

resolve_step_dependencies(consumer_step, available_steps)

Resolve dependencies for a single step.

Parameters:

• consumer_step (str) – Name of the step whose dependencies to resolve

• available_steps (List[str]) – List of available step names

Returns:

Dictionary mapping dependency names to property references

Return type:

Dict[str, PropertyReference]

resolve_with_scoring(consumer_step, available_steps)

Resolve dependencies with detailed compatibility scoring.

Parameters:

• consumer_step (str) – Name of the step whose dependencies to resolve

• available_steps (List[str]) – List of available step names

Returns:

Dictionary with resolved dependencies and detailed scoring information

Return type:

Dict[str, Any]

exception DependencyResolutionError

Bases: Exception

Raised when dependencies cannot be resolved.

create_dependency_resolver(registry=None, semantic_matcher=None)

Create a properly configured dependency resolver.

Parameters:

• registry (SpecificationRegistry | None) – Optional specification registry. If None, creates a new one.

• semantic_matcher (SemanticMatcher | None) – Optional semantic matcher. If None, creates a new one.

Returns:

Configured UnifiedDependencyResolver instance

Return type:

UnifiedDependencyResolver

class SemanticMatcher

Bases: object

Semantic similarity matching for dependency resolution.

calculate_similarity(name1, name2)

Calculate semantic similarity between two names.

Parameters:

• name1 (str) – First name to compare

• name2 (str) – Second name to compare

Returns:

Similarity score between 0.0 and 1.0

Return type:

float

calculate_similarity_with_aliases(name, output_spec)

Calculate semantic similarity between a name and an output specification, considering both logical_name and all aliases.

Parameters:

• name (str) – The name to compare (typically the dependency’s logical_name)

• output_spec (Any) – OutputSpec with logical_name and potential aliases

Returns:

The highest similarity score (0.0 to 1.0) between name and any name in output_spec

Return type:

float

explain_similarity(name1, name2)

Provide detailed explanation of similarity calculation.

Parameters:

• name1 (str) – First name to compare

• name2 (str) – Second name to compare

Returns:

Dictionary with detailed similarity breakdown

Return type:

Dict[str, Any]

find_best_matches(target_name, candidate_names, threshold=0.5)

Find the best matching names from a list of candidates.

Parameters:

• target_name (str) – Name to match against

• candidate_names (List[str]) – List of candidate names

• threshold (float) – Minimum similarity threshold

Returns:

List of (name, score) tuples sorted by score (highest first)

Return type:

List[Tuple[str, float]]

create_pipeline_components(context_name=None)

Create all necessary pipeline components with proper dependencies.

Modules

dependency_resolver	Unified dependency resolver for intelligent pipeline dependency management.
factory	Factory functions for creating pipeline dependency components.
property_reference	Property Reference module for SageMaker property path handling.
registry_manager	Registry manager for coordinating multiple isolated specification registries.
semantic_matcher	Semantic matching utilities for intelligent dependency resolution.
specification_registry	Specification registry for managing step specifications with context isolation.