cursus.steps.configs.config_processing_step_base

Processing Step Base Configuration with Self-Contained Derivation Logic

This module implements the base configuration class for SageMaker Processing steps using a self-contained design where derived fields are private with read-only properties.

class ProcessingStepConfigBase(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.0', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=False, max_runtime_seconds=172800, project_root_folder, processing_instance_count=1, processing_volume_size=500, processing_instance_type_large='ml.m5.4xlarge', processing_instance_type_small='ml.m5.2xlarge', use_large_processing_instance=False, skip_volume_kms=None, processing_source_dir=None, processing_entry_point=None, processing_script_arguments=None, processing_framework_version='1.2-1', **extra_data)

Bases: BasePipelineConfig

Base configuration for SageMaker Processing Steps with self-contained derivation logic.

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

use_large_processing_instance: bool

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_entry_point: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True, 'extra': 'allow', 'protected_namespaces': (), 'validate_assignment': True}

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

property effective_source_dir: str | None

Get effective source directory with hybrid resolution.

Resolution Priority: 1. Hybrid resolution of processing_source_dir 2. Hybrid resolution of source_dir 3. Legacy values (processing_source_dir, source_dir)

property effective_instance_type: str

Get the appropriate instance type based on the use_large_processing_instance flag.

property script_path: str | None

Get script path with hybrid resolution.

Uses modernized effective_source_dir which already includes hybrid resolution.

property resolved_processing_source_dir: str | None

Get resolved processing source directory using hybrid resolution.

get_resolved_script_path()

Get resolved script path for step builders using hybrid resolution.

model_dump(**kwargs)

Override model_dump to include derived properties.

classmethod validate_processing_source_dir(v)

Validate processing source directory format (S3 paths only).

classmethod validate_entry_point_is_relative(v)

Validate entry point is a relative path if provided.

initialize_derived_fields()

Initialize all derived fields once after validation.

validate_entry_point_paths()

Validate entry point configuration requirements (without file existence checks).

get_effective_source_dir()

Get the effective source directory (legacy compatibility).

get_instance_type(size=None)

Get the appropriate instance type based on size parameter or configuration.

Parameters:

size (Optional[str]) – Override ‘small’ or ‘large’. If None, uses use_large_processing_instance.

Returns:

The corresponding instance type

Return type:

str

get_script_path(default_path=None)

Get script path with hybrid resolution and comprehensive fallbacks.

Resolution Priority: 1. Modernized script_path property (includes hybrid resolution) 2. Direct hybrid resolution of entry_point 3. Legacy get_resolved_script_path() method 4. Default path fallback

Parameters:

default_path (str | None) – Default path to use if all resolution methods fail

Returns:

Resolved script path or default_path if not found

Return type:

Optional[str]

get_public_init_fields()

Override get_public_init_fields to include processing-specific fields. Gets a dictionary of public fields suitable for initializing a child config. Includes both base fields (from parent) and processing-specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

model_post_init(context, /)

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

• self (BaseModel) – The BaseModel instance.

• context (Any) – The context.

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

framework_version: str

py_version: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str