cursus.core.base.builder_base

safe_value_for_logging(value)[source]

Safely format a value for logging, handling Pipeline variables appropriately.

Parameters:

value (Any) – Any value that might be a Pipeline variable

Returns:

A string representation safe for logging

Return type:

str

class StepBuilderBase(config, spec=None, sagemaker_session=None, role=None, registry_manager=None, dependency_resolver=None)[source]

Bases: ABC

Base class for all step builders

## Safe Logging Methods

To handle Pipeline variables safely in logs, use these methods:

```python # Instead of: logger.info(f”Using input path: {input_path}”) # May raise TypeError for Pipeline variables

# Use: self.log_info(“Using input path: %s”, input_path) # Handles Pipeline variables safely ```

Standard Pattern for input_names and output_names:

  1. In config classes: `python output_names = {"logical_name": "DescriptiveValue"}  # VALUE used as key in outputs dict input_names = {"logical_name": "ScriptInputName"}    # KEY used as key in inputs dict `

  2. In pipeline code: ```python # Get output using VALUE from output_names output_value = step_a.config.output_names[“logical_name”] output_uri = step_a.properties.ProcessingOutputConfig.Outputs[output_value].S3Output.S3Uri

    # Set input using KEY from input_names inputs = {“logical_name”: output_uri} ```

  3. In step builders: ```python # For outputs - validate using VALUES value = self.config.output_names[“logical_name”] if value not in outputs:

    raise ValueError(f”Must supply an S3 URI for ‘{value}’”)

    # For inputs - validate using KEYS for logical_name in self.config.input_names.keys():

    if logical_name not in inputs:

    raise ValueError(f”Must supply an S3 URI for ‘{logical_name}’”)

    ```

Developers should follow this standard pattern when creating new step builders. The base class provides helper methods to enforce and simplify this pattern:

  • _validate_inputs(): Validates inputs using KEYS from input_names

  • _validate_outputs(): Validates outputs using VALUES from output_names

  • _get_script_input_name(): Maps logical name to script input name

  • _get_output_destination_name(): Maps logical name to output destination name

  • _create_standard_processing_input(): Creates standardized ProcessingInput

  • _create_standard_processing_output(): Creates standardized ProcessingOutput

Property Path Registry:

To bridge the gap between definition-time and runtime, step builders can register property paths that define how to access their outputs at runtime. This solves the issue where outputs are defined statically but only accessible via specific runtime paths.

  • register_property_path(): Registers a property path for a logical output name

  • get_property_paths(): Gets all registered property paths for this step

REGION_MAPPING: Dict[str, str] = {'EU': 'eu-west-1', 'FE': 'us-west-2', 'NA': 'us-east-1'}
STEP_NAME: str | None = None

The canonical registry step name for THIS builder (singular), e.g. “XGBoostTraining”. This is the authoritative identity slot _get_step_name reads (FZ 31e1d3g3 Phase C1): every routed shell + TemplateStepBuilder sets it, the materializer stamps it on synthesized fileless builders, and once the per-step shell classes are deleted it is the ONLY reliable canonical key (the class name collapses to TemplateStepBuilder). Declared here on the root so the base method that reads it owns the slot, and so it is not confused with STEP_NAMES (PLURAL — the whole registry dict, below). None on a hand-written builder, which falls back to the legacy <Name>StepBuilder class-name convention.

property STEP_NAMES: Dict[str, Any]

Lazy load step names with workspace context awareness.

This property now supports workspace-aware step name resolution by: 1. Extracting workspace context from config or environment 2. Using hybrid registry manager for workspace-specific step names 3. Falling back to traditional registry if hybrid is unavailable 4. Maintaining backward compatibility with existing code

Returns:

Step names mapping for the current workspace context

Return type:

Dict[str, str]

COMMON_PROPERTIES = {'dependencies': 'Optional list of dependent steps', 'enable_caching': 'Whether to enable caching for this step (default: True)'}
TRAINING_OUTPUT_PROPERTIES = {'model_data': 'S3 path to the model artifacts', 'model_data_url': 'S3 URL to the model artifacts', 'training_job_name': 'Name of the training job'}
MODEL_OUTPUT_PROPERTIES = {'model': 'SageMaker model object', 'model_artifacts_path': 'S3 path to model artifacts'}
execution_prefix: str | Any | None
get_property_path(logical_name, format_args=None)[source]

Get property path for an output using the specification.

This method retrieves the property path for an output from the specification. It also supports template formatting if format_args are provided.

Parameters:

  • logical_name (str) – Logical name of the output

  • format_args (Dict[str, Any] | None) – Optional dictionary of format arguments for template paths (e.g., {‘output_descriptor’: ‘data’} for paths with placeholders)

Returns:

Property path from specification, formatted with args if provided, or None if not found

Return type:

str | None

get_all_property_paths()[source]

Get all property paths defined in the specification.

Returns:

Mapping from logical output names to runtime property paths

Return type:

dict

log_info(message, *args, **kwargs)[source]

Safely log info messages, handling Pipeline variables.

Parameters:

  • message (str) – The log message

  • *args (Any) – Values to format into the message

  • **kwargs (Any) –

    Values to format into the message

log_debug(message, *args, **kwargs)[source]

Debug version of safe logging

log_warning(message, *args, **kwargs)[source]

Warning version of safe logging

log_error(message, *args, **kwargs)[source]

Error version of safe logging

set_execution_prefix(execution_prefix=None)[source]

Set the execution prefix for dynamic output path resolution.

This method is called by PipelineAssembler to provide the execution prefix that step builders use for dynamic output path generation.

Based on analysis of regional_xgboost.py, only PIPELINE_EXECUTION_TEMP_DIR is used by step builders for output paths. Other pipeline parameters (KMS_ENCRYPTION_KEY_PARAM, VPC_SUBNET, SECURITY_GROUP_ID) are used at the pipeline level, not in step builders.

Parameters:

execution_prefix (str | Any | None) – The execution prefix that can be either: - ParameterString: PIPELINE_EXECUTION_TEMP_DIR from pipeline parameters - str: config.pipeline_s3_loc as fallback - None: No parameter found, will fall back to config.pipeline_s3_loc

validate_configuration()[source]

Validate builder-context configuration requirements (optional hook).

No-op by default. The Pydantic config class is the authority for config validation — required fields, @field_validator / @model_validator constraints, and defaults are all enforced at config construction, BEFORE the builder runs. A config that constructs is valid by definition, so most builders need no override here (FZ 31e1d3e). Override ONLY to assert an invariant the config genuinely cannot express — one involving builder context (self.role / self.session / self.spec / resolved dependencies) or a cross-field rule not yet on the config model.

get_required_dependencies()[source]

Get list of required dependency logical names from specification.

This method provides direct access to the required dependencies defined in the step specification.

Returns:

List of logical names for required dependencies

Raises:

ValueError – If specification is not provided

Return type:

List[str]

get_optional_dependencies()[source]

Get list of optional dependency logical names from specification.

This method provides direct access to the optional dependencies defined in the step specification.

Returns:

List of logical names for optional dependencies

Raises:

ValueError – If specification is not provided

Return type:

List[str]

get_outputs()[source]

Get output specifications directly from the step specification.

This method provides direct access to the outputs defined in the step specification, returning the complete OutputSpec objects.

Returns:

Dictionary mapping output names to their OutputSpec objects

Raises:

ValueError – If specification is not provided

Return type:

Dict[str, Any]

extract_inputs_from_dependencies(dependency_steps)[source]

Extract inputs from dependency steps using the UnifiedDependencyResolver.

Parameters:

dependency_steps (List[Step]) – List of dependency steps

Returns:

Dictionary of inputs extracted from dependency steps

Raises:

ValueError – If dependency resolver is not available or specification is not provided

Return type:

Dict[str, Any]

abstractmethod create_step(**kwargs)[source]

Create pipeline step.

This method should be implemented by all step builders to create a SageMaker pipeline step. It accepts a dictionary of keyword arguments that can be used to configure the step.

Common parameters that all step builders should handle: - dependencies: Optional list of steps that this step depends on - enable_caching: Whether to enable caching for this step (default: True)

Step-specific parameters should be extracted from kwargs as needed.

Parameters:

**kwargs (Any) – Keyword arguments for configuring the step

Returns:

SageMaker pipeline step

Return type:

Step