cursus.steps.configs

Step configurations module.

This module contains configuration classes for all pipeline steps, providing type-safe configuration management with validation and serialization capabilities.

class BasePipelineConfig(*, 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, **extra_data)

Bases: BaseModel, ABC

Base configuration with shared pipeline attributes and self-contained derivation logic.

property aws_region: str

Get AWS region based on region code.

categorize_fields()

Categorize all fields into three tiers: 1. Tier 1: Essential User Inputs - public fields with no defaults (required) 2. Tier 2: System Inputs - public fields with defaults (optional) 3. Tier 3: Derived Fields - properties that access private attributes

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names

Return type:

Dict[str, List[str]]

property effective_source_dir: str | None

Get effective source directory with hybrid resolution.

This base implementation works with just source_dir (which can be None). Processing configs override this to handle both processing_source_dir and source_dir.

Resolution Priority: 1. Hybrid resolution of source_dir 2. Legacy value (source_dir) 3. None if source_dir is not provided

classmethod from_base_config(base_config, **kwargs)

Create a new configuration instance from a base configuration. This is a virtual method that all derived classes can use to inherit from a parent config.

Parameters:

• base_config (BasePipelineConfig) – Parent BasePipelineConfig instance

• **kwargs (Any) – Additional arguments specific to the derived class

Returns:

A new instance of the derived class initialized with parent fields and additional kwargs

Return type:

BasePipelineConfig

classmethod get_config_class_name(step_name)

Get the configuration class name from a step name using existing registry functions.

get_environment_variables(declared_env_vars=None)

Resolve the env-var VALUES for the names the step interface DECLARES (FZ 31e1d3g).

The interface (.step.yaml env_vars) declares WHICH env vars a step uses; this config supplies the VALUES. The single-source rule: the interface drives the key set, config resolves each one — by _env_overrides() first (computed/aliased), else convention (NAME -> self.name, type-formatted). Names that resolve to nothing are omitted, so a declared-but-absent optional falls back to its interface default in the builder.

Subclasses with a bespoke collector may override this entirely (and ignore declared_env_vars) to return their full env dict — that path is preserved for back-compat.

get_job_arguments()

Build the script’s CLI arguments — config is the single source (FZ 31e1d3h).

Base default: no CLI arguments (None). The builder’s _get_job_arguments calls this; a step-config that passes arguments to its script OVERRIDES this method. The common --job_type case has a ready helper, _job_type_arg() — a config opts in with return self._job_type_arg(). None (not []) is the “no args” contract the SDK ProcessingStep / processor.run expect.

get_public_init_fields()

Get a dictionary of public fields suitable for initializing a child config. Only includes fields that should be passed to child class constructors. Both essential user inputs and system inputs with defaults or user-overridden values are included to ensure all user customizations are properly propagated to derived classes.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_contract()

Get script contract for this configuration using optimized step catalog discovery.

This optimized implementation uses the step catalog system for efficient contract discovery with O(1) lookups and workspace awareness, falling back to legacy methods for backward compatibility.

Returns:

Script contract instance or None if not available

Return type:

Any | None

get_script_path(default_path=None)

Get script path for this configuration.

This method provides a default implementation that returns None, since not all step types require scripts (e.g., Model creation steps don’t need scripts).

Derived classes that need script paths should override this method with their specific requirements: - Processing steps: Combine entry_point with source_dir - Training steps: Use contract entry_point or combine with source_dir - Model steps: May not need scripts at all

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path, default_path, or None if not applicable for this step type

Return type:

str | None

classmethod get_step_name(config_class_name)

Get the step name for a configuration class using existing registry functions.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property pipeline_description: str

Get pipeline description derived from service_name, model_class, and region.

property pipeline_name: str

Get pipeline name derived from author, service_name, model_class, and region.

property pipeline_s3_loc: str

Get S3 location for pipeline artifacts.

print_config()

Print complete configuration information organized by tiers. This method automatically categorizes fields by examining their characteristics: - Tier 1: Essential User Inputs (public fields without defaults) - Tier 2: System Inputs (public fields with defaults) - Tier 3: Derived Fields (properties that provide access to private fields)

resolve_hybrid_path(relative_path)

Resolve a path using the hybrid path resolution system.

This method uses the hybrid path resolution system to find files across different deployment scenarios (Lambda/MODS bundled, development monorepo, pip-installed separated).

Parameters:

relative_path (str) – Relative path from project root to target

Returns:

Resolved absolute path if found, None otherwise

Return type:

str | None

property resolved_source_dir: str | None

Get resolved source directory using hybrid resolution.

Returns None if source_dir is not provided, since it’s optional in base class. Processing, training, and model step configs should ensure source_dir is provided.

property script_contract: Any | None

Property accessor for script contract.

Returns:

Script contract instance or None if not available

property step_catalog: Any | None

Lazy-loaded step catalog instance for optimized component discovery.

Returns:

StepCatalog instance or None if initialization fails

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

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.

property effective_instance_type: str

Get the appropriate instance type based on the use_large_processing_instance flag.

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)

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

get_resolved_script_path()

Get resolved script path for step builders using hybrid resolution.

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]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property resolved_processing_source_dir: str | None

Get resolved processing source directory using hybrid resolution.

property script_path: str | None

Get script path with hybrid resolution.

Uses modernized effective_source_dir which already includes hybrid resolution.

classmethod validate_entry_point_is_relative(v)

Validate entry point is a relative path if provided.

validate_entry_point_paths()

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

classmethod validate_processing_source_dir(v)

Validate processing source directory format (S3 paths only).

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

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

class ActiveSampleSelectionConfig(*, 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='active_sample_selection.py', processing_script_arguments=None, processing_framework_version='1.2-1', use_case, id_field, label_field, score_field, selection_strategy='confidence_threshold', output_format='csv', confidence_threshold=0.9, k_per_class=100, max_samples=0, uncertainty_mode='margin', batch_size=32, metric='euclidean', random_seed=42, score_field_prefix='prob_class_', job_type='ssl_selection', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Active Sample Selection step.

Supports both Semi-Supervised Learning (SSL) and Active Learning workflows with Pydantic validation to prevent strategy misuse.

Three-Tier Configuration: - Tier 1: Essential User Inputs (none - all have defaults) - Tier 2: System Fields with Defaults (all selection parameters) - Tier 3: Derived Fields (inherited from ProcessingStepConfigBase)

get_environment_variables()

Get environment variables for the processing job.

Returns:

Dictionary of environment variables

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include selection-specific fields. Gets a dictionary of public fields suitable for initializing a child config.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

classmethod validate_output_format(v)

Validate output format (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_strategy(v)

Validate selection strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

validate_strategy_use_case_compatibility()

⚠️ CRITICAL: Validate strategy is compatible with use case.

This cross-field validation prevents: - Using uncertainty strategies for SSL (creates noisy pseudo-labels) - Using confidence strategies for Active Learning (wastes human effort)

Raises:

ValueError – If strategy is incompatible with use_case

classmethod validate_use_case(v)

Validate use case is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

use_case: Literal['ssl', 'active_learning', 'auto']

id_field: str

label_field: str

score_field: str | None

selection_strategy: Literal['confidence_threshold', 'top_k_per_class', 'uncertainty', 'diversity', 'badge']

output_format: Literal['csv', 'parquet']

confidence_threshold: float

k_per_class: int

max_samples: int

uncertainty_mode: Literal['margin', 'entropy', 'least_confidence']

batch_size: int

metric: Literal['euclidean', 'cosine']

random_seed: int

score_field_prefix: str

processing_entry_point: str

processing_framework_version: str

job_type: str

use_large_processing_instance: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

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

class BatchTransformStepConfig(*, 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, job_type, transform_instance_type='ml.m5.large', transform_instance_count=1, content_type='text/csv', accept='text/csv', split_type='Line', assemble_with='Line', input_filter='$[1:]', output_filter='$[-1]', join_source='Input', **extra_data)

Bases: BasePipelineConfig

Configuration for a generic SageMaker BatchTransform step. Inherits all the BasePipelineConfig attributes (bucket, region, etc.) and adds just what’s needed to drive a TransformStep.

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

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.

validate_config()

Validate join and assemble configurations.

job_type: str

transform_instance_type: str

transform_instance_count: int

content_type: str

accept: str

split_type: str

assemble_with: str | None

input_filter: str | None

output_filter: str | None

join_source: str

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

class BedrockBatchProcessingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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='bedrock_batch_processing.py', processing_script_arguments=None, processing_framework_version='1.2-1', bedrock_batch_role_arn, job_type='training', bedrock_primary_model_id='anthropic.claude-sonnet-4-5-20250929-v1:0', bedrock_fallback_model_id='anthropic.claude-sonnet-4-20250514-v1:0', bedrock_inference_profile_arn=None, bedrock_inference_profile_required_models=<factory>, bedrock_max_tokens=32768, bedrock_temperature=1.0, bedrock_top_p=0.999, bedrock_batch_size=10, bedrock_max_retries=3, bedrock_output_column_prefix='llm_', bedrock_skip_error_records=False, bedrock_concurrency_mode='sequential', bedrock_max_concurrent_workers=5, bedrock_rate_limit_per_second=10, bedrock_batch_mode='auto', bedrock_batch_threshold=1000, bedrock_batch_timeout_hours=24, bedrock_max_records_per_job=45000, bedrock_max_concurrent_batch_jobs=20, bedrock_max_input_field_length=400000, bedrock_truncation_enabled=True, bedrock_log_truncations=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Bedrock Batch Processing step using three-tier design.

This step processes input data through AWS Bedrock models using batch inference capabilities with automatic fallback to real-time processing. Integrates with generated prompt templates and validation schemas from the Bedrock Prompt Template Generation step. Provides cost-efficient processing for large datasets.

Tier 1: Essential user inputs (required) Tier 2: System inputs with defaults (optional) Tier 3: Derived fields (private with property access)

property batch_configuration: Dict[str, Any]

Get batch processing configuration details.

property bedrock_environment_variables: Dict[str, str]

Get environment variables for the Bedrock batch processing step.

property cost_optimization_info: Dict[str, Any]

Get cost optimization information.

property effective_inference_profile_required_models: List[str]

Get effective list of models requiring inference profiles with auto-detection.

get_batch_processing_info()

Get detailed batch processing configuration information.

Returns:

Batch processing details and recommendations

Return type:

Dict[str, Any]

get_environment_variables()

Get all environment variables for the step builder.

Returns:

Complete environment variables dictionary

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h). job_type + batch/retry overrides.

get_performance_estimate()

Get estimated performance characteristics including batch processing.

Returns:

Performance estimates and recommendations

Return type:

Dict[str, Any]

get_public_init_fields()

Override get_public_init_fields to include Bedrock batch-specific fields. Gets a dictionary of public fields suitable for initializing a child config.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_path(default_path=None)

Get script path for the Bedrock batch processing step.

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path resolved from processing_entry_point and source directories

Return type:

str | None

initialize_derived_fields()

Initialize all derived fields once after validation.

is_production_ready()

Check if configuration is production-ready.

Returns:

True if configuration has production-ready settings

Return type:

bool

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property processing_metadata: Dict[str, Any]

Get processing step metadata.

classmethod validate_batch_mode(v)

Validate batch processing mode.

classmethod validate_batch_role_arn(v)

Validate batch role ARN format.

classmethod validate_concurrency_mode(v)

Validate concurrency mode.

classmethod validate_inference_profile_models(v)

Validate inference profile required models list.

classmethod validate_job_type(v)

Validate job_type is one of the allowed values.

classmethod validate_primary_model_id(v)

Validate primary model ID format.

validate_production_readiness()

Validate configuration for production readiness.

bedrock_batch_role_arn: str

job_type: str

bedrock_primary_model_id: str

bedrock_fallback_model_id: str | None

bedrock_inference_profile_arn: str | None

bedrock_inference_profile_required_models: List[str]

bedrock_max_tokens: int

bedrock_temperature: float

bedrock_top_p: float

bedrock_batch_size: int

bedrock_max_retries: int

bedrock_output_column_prefix: str

bedrock_skip_error_records: bool

bedrock_concurrency_mode: str

bedrock_max_concurrent_workers: int

bedrock_rate_limit_per_second: int

bedrock_batch_mode: str

bedrock_batch_threshold: int

bedrock_batch_timeout_hours: int

bedrock_max_records_per_job: int

bedrock_max_concurrent_batch_jobs: int

bedrock_max_input_field_length: int

bedrock_truncation_enabled: bool

bedrock_log_truncations: bool

processing_entry_point: str

framework_version: str

py_version: str

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_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class BedrockProcessingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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='bedrock_processing.py', processing_script_arguments=None, processing_framework_version='1.2-1', bedrock_inference_profile_arn, job_type='training', bedrock_primary_model_id='anthropic.claude-sonnet-4-5-20250929-v1:0', bedrock_fallback_model_id='anthropic.claude-sonnet-4-20250514-v1:0', bedrock_inference_profile_required_models=<factory>, bedrock_max_tokens=32000, bedrock_temperature=1.0, bedrock_top_p=0.999, bedrock_batch_size=10, bedrock_max_retries=3, bedrock_output_column_prefix='llm_', bedrock_skip_error_records=False, bedrock_concurrency_mode='sequential', bedrock_max_concurrent_workers=5, bedrock_rate_limit_per_second=10, bedrock_max_input_field_length=400000, bedrock_truncation_enabled=True, bedrock_log_truncations=True, bedrock_user_prompt_template=None, bedrock_system_prompt=None, bedrock_input_placeholders=None, bedrock_validation_schema=None, bedrock_use_structured_output=False, bedrock_use_converse_api=False, bedrock_adaptive_rate_limiting=False, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Bedrock Processing step using three-tier design.

This step processes input data through AWS Bedrock models using generated prompt templates and validation schemas from the Bedrock Prompt Template Generation step. Supports template-driven response processing with dynamic Pydantic model creation and both sequential and concurrent processing modes.

Tier 1: Essential user inputs (required) Tier 2: System inputs with defaults (optional) Tier 3: Derived fields (private with property access)

property bedrock_environment_variables: Dict[str, str]

Get environment variables for the Bedrock processing step.

property concurrency_configuration: Dict[str, Any]

Get concurrency configuration details.

property effective_inference_profile_required_models: List[str]

Get effective list of models requiring inference profiles with auto-detection.

get_environment_variables()

Get all environment variables for the step builder.

Returns:

Complete environment variables dictionary

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h). job_type + batch/retry overrides.

get_performance_estimate()

Get estimated performance characteristics.

Returns:

Performance estimates and recommendations

Return type:

Dict[str, Any]

get_public_init_fields()

Override get_public_init_fields to include Bedrock-specific fields. Gets a dictionary of public fields suitable for initializing a child config.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_path(default_path=None)

Get script path for the Bedrock processing step.

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path resolved from processing_entry_point and source directories

Return type:

str | None

initialize_derived_fields()

Initialize all derived fields once after validation.

is_production_ready()

Check if configuration is production-ready.

Returns:

True if configuration has production-ready settings

Return type:

bool

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property processing_metadata: Dict[str, Any]

Get processing step metadata.

classmethod validate_concurrency_mode(v)

Validate concurrency mode.

classmethod validate_inference_profile_models(v)

Validate inference profile required models list.

classmethod validate_job_type(v)

Validate job_type is one of the allowed values.

classmethod validate_primary_model_id(v)

Validate primary model ID format.

validate_production_readiness()

Validate configuration for production readiness.

bedrock_inference_profile_arn: str

job_type: str

bedrock_primary_model_id: str

bedrock_fallback_model_id: str | None

bedrock_inference_profile_required_models: List[str]

bedrock_max_tokens: int

bedrock_temperature: float

bedrock_top_p: float

bedrock_batch_size: int

bedrock_max_retries: int

bedrock_output_column_prefix: str

bedrock_skip_error_records: bool

bedrock_concurrency_mode: str

bedrock_max_concurrent_workers: int

bedrock_rate_limit_per_second: int

bedrock_max_input_field_length: int

bedrock_truncation_enabled: bool

bedrock_log_truncations: bool

bedrock_user_prompt_template: str | None

bedrock_system_prompt: str | None

bedrock_input_placeholders: List[str] | None

bedrock_validation_schema: Dict[str, Any] | None

bedrock_use_structured_output: bool

bedrock_use_converse_api: bool

bedrock_adaptive_rate_limiting: bool

processing_entry_point: str

framework_version: str

py_version: str

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_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class BedrockPromptTemplateGenerationConfig(*, 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='bedrock_prompt_template_generation.py', processing_script_arguments=None, processing_framework_version='1.2-1', input_placeholders, prompt_configs_path='prompt_configs', template_task_type='classification', template_style='structured', validation_level='standard', output_format_type='structured_json', required_output_fields=['category', 'confidence', 'key_evidence', 'reasoning'], include_examples=True, generate_validation_schema=True, template_version='1.0', system_prompt_settings=None, output_format_settings=None, instruction_settings=None, category_definitions=None, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Bedrock Prompt Template Generation step using three-tier design.

This step generates structured prompt templates for classification tasks using the 5-component architecture pattern optimized for LLM performance.

Tier 1: Essential user inputs (required) Tier 2: System inputs with defaults (optional) Tier 3: Derived fields (private with property access)

property effective_categories: CategoryDefinitionList | None

Get effective category definitions from direct definitions.

property effective_instruction_config: Dict[str, Any]

Get instruction configuration from typed settings or defaults.

property effective_output_format_config: Dict[str, Any]

Get output format configuration from typed settings or defaults.

property effective_system_prompt_config: Dict[str, Any]

Get system prompt configuration from typed settings or defaults.

property environment_variables: Dict[str, str]

Get environment variables for the processing step (file-based approach).

generate_prompt_config_bundle()

Generate complete prompt configuration bundle for the refactored file-based approach.

Creates JSON files needed by the script in the configured prompt_configs_path: - system_prompt.json (only if system_prompt_settings is provided) - output_format.json (only if output_format_settings is provided) - instruction.json (only if instruction_settings is provided) - category_definitions.json (only if category_definitions is provided)

All configuration files are optional - the script will use defaults when files are missing.

Raises:

ValueError – If prompt_configs_path is not configured

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h). Boolean flags + template version.

get_public_init_fields()

Override get_public_init_fields to include template-specific fields. Gets a dictionary of public fields suitable for initializing a child config.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_path(default_path=None)

Get script path for the Bedrock prompt template generation step.

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path resolved from processing_entry_point and source directories

Return type:

str | None

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property resolved_prompt_configs_path: str | None

Get resolved absolute path for prompt configurations with hybrid resolution.

Uses effective_source_dir from base class for consistency.

Returns:

Absolute path to prompt configs directory, or None if not configured

Raises:

ValueError – If prompt_configs_path is set but source directory cannot be resolved

property template_metadata: Dict[str, Any]

Get template generation metadata.

input_placeholders: List[str]

prompt_configs_path: str

template_task_type: str

template_style: str

validation_level: str

output_format_type: str

required_output_fields: List[str]

include_examples: bool

generate_validation_schema: bool

template_version: str

system_prompt_settings: SystemPromptConfig | None

output_format_settings: OutputFormatConfig | None

instruction_settings: InstructionConfig | None

category_definitions: List[CategoryDefinition] | None

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class CurrencyConversionConfig(*, 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='currency_conversion.py', processing_script_arguments=None, processing_framework_version='1.2-1', currency_conversion_vars, currency_conversion_dict, job_type='training', currency_code_field=None, marketplace_id_field=None, default_currency='USD', n_workers=50, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Currency Conversion step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

This configuration follows the specification-driven approach where inputs and outputs are defined by step specifications and script contracts, not by hardcoded dictionaries.

property environment_variables: Dict[str, str]

Generate environment variables for the currency conversion script.

Returns:

Dictionary of environment variables

get_environment_variables(declared_env_vars=None)

Generate the currency-conversion env vars (the single env source; FZ 31e1d3g).

declared_env_vars is accepted for the builder’s names-driven contract but ignored — these values are all computed (JSON-encoded conversion vars/dict), so this config returns its full computed env dict regardless of the declared name set.

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include currency conversion specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

validate_currency_fields()

Ensure at least one of currency_code_field or marketplace_id_field is not None.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_job_type(v)

Validate job_type is lowercase alphanumeric (with underscores) — matches the peers’ validator (RiskTableMapping/Tabular) and the script’s choices.

classmethod validate_vars(v)

Validate currency conversion variables list.

currency_conversion_vars: List[str]

currency_conversion_dict: CurrencyConversionDictConfig

job_type: str

processing_entry_point: str

currency_code_field: str | None

marketplace_id_field: str | None

default_currency: str

n_workers: int

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_script_arguments: List[str] | None

processing_framework_version: str

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

class DataUploadingConfig(*, 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, provider_name, table_name, table_version, cradle_account, schema_sdl, job_type=None, input_s3_location=None, input_format='CSV', has_header=True, cluster_type='MEDIUM', instance_type='ml.m5.xlarge', instance_count=1, volume_size_in_gb=30, **extra_data)

Bases: BasePipelineConfig

Configuration for the DataUploading step (S3 → Andes via SAIS SDK).

property job_config: Dict[str, Any]

Assemble the full job config dict for CreateDataUploadJobRequest.

This is what gets written to /opt/ml/processing/config/config and deserialized via CreateDataUploadJobRequest.__from_coral__(**config). Note: inputS3Url is NOT included here — it’s injected at runtime via –input-path argument by the SDK processor.

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

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.

validate_cluster_type()

Validate cluster_type is a known value.

provider_name: str

table_name: str

table_version: int

cradle_account: str

schema_sdl: str

job_type: str | None

input_s3_location: str | None

input_format: str

has_header: bool

cluster_type: str

instance_type: str

instance_count: int

volume_size_in_gb: int

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

class CradleDataLoadingConfig(*, 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, job_type, data_sources_spec, transform_spec, output_spec, cradle_job_spec, s3_input_override=None, **extra_data)

Bases: BasePipelineConfig

Top‐level configuration for creating a CreateCradleDataLoadJobRequest with three-tier field classification.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

This class inherits from BasePipelineConfig (not BaseCradleComponentConfig) to maintain consistency with other pipeline configurations.

categorize_fields()

Categorize all fields into three tiers: 1. Tier 1: Essential User Inputs - fields with no defaults (required) 2. Tier 2: System Inputs - fields with defaults (optional) 3. Tier 3: Derived Fields - properties that access private attributes

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names

Return type:

Dict[str, List[str]]

check_split_and_override()

Check consistency of split settings and overrides.

get_all_tiered_fields()

Get a flattened list of all fields (including nested fields) organized by tier.

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names with dot notation for nesting

Return type:

Dict[str, List[str]]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

classmethod validate_job_type(v)

job_type: str

data_sources_spec: DataSourcesSpecificationConfig

transform_spec: TransformSpecificationConfig

output_spec: OutputSpecificationConfig

cradle_job_spec: CradleJobSpecificationConfig

s3_input_override: str | None

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

class DummyDataLoadingConfig(*, 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='dummy_data_loading.py', processing_script_arguments=None, processing_framework_version='1.2-1', data_source, job_type='training', max_file_size_mb=1000, supported_formats=['csv', 'parquet', 'json', 'jsonl'], write_data_shards=False, shard_size=10000, output_format='CSV', max_workers=0, batch_size=5, optimize_memory=False, streaming_batch_size=0, enable_true_streaming=False, metadata_format='JSON', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for a dummy data loading step.

This configuration follows the three-tier field categorization: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that users can override 3. Tier 3: Derived Fields - fields calculated from other fields, stored in private attributes

get_data_source_uri()

Get the data source as a string URI.

Returns:

Data source URI (S3 or local path)

Return type:

str

get_environment_variables()

Get environment variables for the processing script.

Returns:

Dictionary of environment variables to be passed to the processing script.

Return type:

dict

get_supported_extensions()

Get supported file extensions based on configured formats.

Returns:

Set of supported file extensions (with dots)

Return type:

set[str]

is_s3_source()

Check if the data source is an S3 URI.

Returns:

True if data source is S3, False if local path

Return type:

bool

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

validate_config()

Validate configuration and ensure defaults are set.

This validator ensures that: 1. Data source is provided and properly formatted 2. Entry point is provided 3. Script contract is available and valid 4. Required input/output paths are defined in the script contract

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_metadata_format(v)

Ensure metadata_format is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical uppercase allowed value.

classmethod validate_output_format(v)

Ensure output_format is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical uppercase allowed value.

data_source: str | Path

processing_entry_point: str

job_type: str

max_file_size_mb: int

supported_formats: list[str]

write_data_shards: bool

shard_size: int

output_format: str

max_workers: int

batch_size: int

optimize_memory: bool

streaming_batch_size: int

enable_true_streaming: bool

metadata_format: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class DummyTrainingConfig(*, 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='dummy_training.py', processing_script_arguments=None, processing_framework_version='1.2-1', pretrained_model_path, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for DummyTraining step with flexible input modes.

This configuration follows the Three-Tier Config Design pattern:

Tier 1 (Essential Fields): Required user inputs - pretrained_model_path: Path to model artifacts (S3 URI, local path, or None) - Inherited required fields from BasePipelineConfig and ProcessingStepConfigBase

Tier 2 (System Fields): Fields with defaults that can be overridden - processing_entry_point: Entry point script (default: “dummy_training.py”) - Instance types, volume sizes, etc. (inherited from base)

Tier 3 (Derived Fields): Calculated from Tier 1 and Tier 2 - Inherited derived fields like effective_source_dir, script_path

## Model Artifacts Input (Tier 1 Essential Field)

The pretrained_model_path field accepts three types of values:

1. None (default) - SOURCE Fallback Assumption: - Assumes model.tar.gz is located at source_dir/models/model.tar.gz - Default behavior for backward compatibility - Example: pretrained_model_path=None or omit the field

2. S3 URI - Explicit S3 Path: - Full S3 path to model directory or file - Examples:

   • s3://my-bucket/models/ (directory)

   • s3://my-bucket/models/model.tar.gz (file)

3. Local Path - Explicit Local Directory: - Relative or absolute path to model directory - Examples:

   • ./models/ (relative)

   • /absolute/path/to/models/ (absolute)

   • source_dir/models/ (relative to source)

Priority Resolution: Config field → Dependency injection → SOURCE fallback

## Hyperparameters Resolution

Hyperparameters follow dependency injection pattern: - From hyperparameters_s3_uri channel (if provided via dependency injection) - Falls back to multiple SOURCE locations:

• /opt/ml/code/hyperparams/hyperparameters.json

• source_dir/hyperparams/hyperparameters.json

• source_dir/hyperparameters.json

## Use Cases

• SOURCE Fallback: pretrained_model_path=None (default, uses source_dir/models/)

• Explicit S3: pretrained_model_path=”s3://bucket/path/to/models/”

• Explicit Local: pretrained_model_path=”path/to/models/” or “./models/”

• Absolute Path: pretrained_model_path=”/absolute/path/to/models/”

## Expected Source Directory Structure (when pretrained_model_path=None)

``` source_dir/ ├── dummy_training.py # Main processing script ├── models/ # Model directory │ └── model.tar.gz # Pre-trained model artifacts └── hyperparams/ # Hyperparameters directory (optional)

└── hyperparameters.json # Hyperparameters file

```

## Example Configs

```python # SOURCE fallback (None assumption) config = DummyTrainingConfig(

pretrained_model_path=None, # or omit entirely # … other required fields

)

# Explicit S3 path config = DummyTrainingConfig(

pretrained_model_path=”s3://my-bucket/models/”, # … other required fields

)

# Explicit local path config = DummyTrainingConfig(

pretrained_model_path=”./models/”, # … other required fields

)

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

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.

validate_config()

Validate configuration for INTERNAL node with optional inputs.

For INTERNAL nodes with optional dependencies, we validate: - Entry point is specified - Script contract is valid - Output paths are correctly defined

File existence is checked at runtime, not configuration time.

Returns:

Self with validated configuration

Return type:

DummyTrainingConfig

classmethod validate_pretrained_model_path(v)

Validate pretrained_model_path is None, S3 URI, or local directory path.

Converts Path objects to strings for consistent handling.

Parameters:

v (str | Path | None) – Path value to validate (str, Path, or None)

Returns:

Validated path as string or None

Raises:

ValueError – If path format is invalid

Return type:

str | None

pretrained_model_path: str | Path | None

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class FeatureSelectionConfig(*, 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='feature_selection.py', processing_script_arguments=None, processing_framework_version='1.2-1', label_field, job_type='training', feature_selection_methods='variance, correlation, mutual_info, rfe', n_features_to_select=10, correlation_threshold=0.95, variance_threshold=0.01, random_state=42, combination_strategy='voting', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Feature Selection step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

property environment_variables: Dict[str, str]

Get environment variables for the feature selection script.

Returns:

Dictionary of environment variables

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include feature selection specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

property method_list: List[str]

Get list of feature selection methods from comma-separated string.

Returns:

List of method names

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_combination_strategy(v)

Ensure combination_strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_feature_selection_methods(v)

Validate feature selection methods string.

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_label_field(v)

Ensure label_field is a non-empty string.

label_field: str

processing_entry_point: str

job_type: str

feature_selection_methods: str

n_features_to_select: int

correlation_threshold: float

variance_threshold: float

random_state: int

combination_strategy: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class LabelRulesetExecutionConfig(*, 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='label_ruleset_execution.py', processing_script_arguments=None, processing_framework_version='1.2-1', job_type, fail_on_missing_fields=True, enable_rule_match_tracking=True, enable_progress_logging=True, preferred_input_format='', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Label Ruleset Execution step using three-tier design.

This step applies validated rulesets to processed data to generate classification labels using priority-based rule evaluation with execution-time field validation. Supports stacked preprocessing patterns by using processed_data for both input and output.

Tier 1: Essential user inputs (required) Tier 2: System inputs with defaults (optional) Tier 3: Derived fields (private with property access)

property execution_configuration: Dict[str, Any]

Get execution configuration details.

property execution_environment_variables: Dict[str, str]

Get environment variables for the label ruleset execution step.

get_environment_variables()

Get all environment variables for the step builder.

Returns:

Complete environment variables dictionary

Return type:

Dict[str, str]

get_execution_info()

Get detailed execution configuration information.

Returns:

Execution details and recommendations

Return type:

Dict[str, Any]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include execution-specific fields. Gets a dictionary of public fields suitable for initializing a child config.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_path(default_path=None)

Get script path for the label ruleset execution step.

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path resolved from processing_entry_point and source directories

Return type:

str | None

initialize_derived_fields()

Initialize all derived fields once after validation.

is_production_ready()

Check if configuration is production-ready.

Returns:

True if configuration has production-ready settings

Return type:

bool

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property processing_metadata: Dict[str, Any]

Get processing step metadata.

classmethod validate_job_type(v)

Validate job_type is one of the allowed values.

classmethod validate_preferred_input_format(v)

Validate preferred_input_format is one of the allowed values.

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value. The consuming script lowercases PREFERRED_INPUT_FORMAT on read, so canonical case is not load-bearing.

validate_production_readiness()

Validate configuration for production readiness.

job_type: str

fail_on_missing_fields: bool

enable_rule_match_tracking: bool

enable_progress_logging: bool

preferred_input_format: str

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class LabelRulesetGenerationConfig(*, 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='label_ruleset_generation.py', processing_script_arguments=None, processing_framework_version='1.2-1', label_config, rule_definitions, ruleset_configs_path='ruleset_configs', enable_field_validation=True, enable_label_validation=True, enable_logic_validation=True, enable_rule_optimization=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Label Ruleset Generation step using three-tier design.

This step validates and optimizes user-defined classification rules for transparent, maintainable rule-based label mapping in ML training pipelines.

Tier 1: Essential user inputs (required) Tier 2: System inputs with defaults (optional) Tier 3: Derived fields (private with property access)

property environment_variables: Dict[str, str]

Get environment variables for the processing step.

generate_ruleset_config_bundle()

Generate complete ruleset configuration bundle.

Creates JSON files for non-None configurations in the configured ruleset_configs_path: - label_config.json (if label_config is not None) - field_config.json (if field_config is not None) - ruleset.json (if rule_definitions is not None)

Only generates files for configurations that are provided.

Raises:

ValueError – If ruleset_configs_path is not configured

get_public_init_fields()

Override get_public_init_fields to include ruleset-specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

get_script_path(default_path=None)

Get script path for the label ruleset generation step.

Parameters:

default_path (str | None) – Default script path to use if not found via other methods

Returns:

Script path resolved from processing_entry_point and source directories

Return type:

str | None

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property resolved_ruleset_configs_path: str | None

Get resolved absolute path for ruleset configurations.

Uses effective_source_dir from base class for consistency.

Returns:

Absolute path to ruleset configs directory, or None if not configured

Raises:

ValueError – If ruleset_configs_path is set but source directory cannot be resolved

label_config: LabelConfig

rule_definitions: RulesetDefinitionList

ruleset_configs_path: str

enable_field_validation: bool

enable_label_validation: bool

enable_logic_validation: bool

enable_rule_optimization: bool

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class LightGBMModelEvalConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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=True, skip_volume_kms=None, processing_source_dir=None, processing_entry_point='lightgbm_model_eval.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', eval_metric_choices=<factory>, comparison_mode=False, previous_score_field='', comparison_metrics='all', statistical_tests=True, comparison_plots=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for LightGBM model evaluation step with self-contained derivation logic.

This class defines the configuration parameters for the LightGBM model evaluation step, which calculates evaluation metrics for trained models. This is crucial for measuring model performance and comparing different models or configurations.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the model evaluation script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_eval_config()

Additional validation specific to evaluation configuration

id_name: str

label_name: str

processing_entry_point: str

job_type: str

eval_metric_choices: List[str]

framework_version: str

py_version: str

use_large_processing_instance: bool

comparison_mode: bool

previous_score_field: str

comparison_metrics: str

statistical_tests: bool

comparison_plots: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class LightGBMModelInferenceConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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='lightgbm_model_inference.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', output_format='csv', json_orient='records', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for LightGBM model inference step with self-contained derivation logic.

This class defines the configuration parameters for the LightGBM model inference step, which generates predictions from trained models without computing evaluation metrics. This is designed for pure inference workflows where predictions are needed for downstream processing (e.g., model calibration, batch scoring).

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the model inference script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_inference_config()

Additional validation specific to inference configuration

classmethod validate_json_orient(v)

Validate JSON orientation is supported.

classmethod validate_output_format(v)

Validate output format is supported.

id_name: str

label_name: str

processing_entry_point: str

job_type: str

output_format: str

json_orient: str

framework_version: str

py_version: str

use_large_processing_instance: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class LightGBMTrainingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='lightgbm', current_date=<factory>, framework_version='2.1.2', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=True, max_runtime_seconds=172800, project_root_folder, training_entry_point, training_instance_type='ml.m5.4xlarge', training_instance_count=1, training_volume_size=30, ca_repository_arn='arn:aws:codeartifact:us-west-2:149122183214:repository/amazon/secure-pypi', skip_hyperparameters_s3_uri=True, use_precomputed_imputation=False, use_precomputed_risk_tables=False, use_precomputed_features=False, use_native_categorical=True, job_type=None, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker LightGBM Training Step. This version is streamlined to work with specification-driven architecture. Input/output paths are now provided via step specifications and dependencies.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the LightGBM training script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

property hyperparameter_file: str

Get hyperparameter file path.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_job_type(v)

Validate job_type is open (any lowercase alphanumeric with underscores; None = standard).

training_entry_point: str

training_instance_type: str

training_instance_count: int

training_volume_size: int

framework_version: str

py_version: str

ca_repository_arn: str

model_class: str

skip_hyperparameters_s3_uri: bool

use_secure_pypi: bool

use_precomputed_imputation: bool

use_precomputed_risk_tables: bool

use_precomputed_features: bool

use_native_categorical: bool

job_type: str | None

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

current_date: str

source_dir: str | None

enable_caching: bool

max_runtime_seconds: int

project_root_folder: str

class LightGBMMTTrainingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='lightgbmmt', current_date=<factory>, framework_version='2.1.2', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=True, max_runtime_seconds=172800, project_root_folder, training_entry_point, training_instance_type='ml.m5.4xlarge', training_instance_count=1, training_volume_size=30, ca_repository_arn='arn:aws:codeartifact:us-west-2:149122183214:repository/amazon/secure-pypi', max_run_seconds=86400, skip_hyperparameters_s3_uri=True, use_precomputed_imputation=False, use_precomputed_risk_tables=False, use_precomputed_features=False, use_native_categorical=True, job_type=None, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker LightGBMMT Training Step.

Uses custom LightGBMMT Docker image for multi-task learning with: - Shared tree structures across related tasks - Adaptive task weighting based on similarity (JS divergence) - Knowledge distillation for performance stabilization - Refactored loss functions and model architecture

Hyperparameters are managed separately via LightGBMMtModelHyperparameters and saved as hyperparameters.json in the source_dir.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the LightGBMMT training script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

property hyperparameter_file: str

Get hyperparameter file path.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_job_type(v)

Validate job_type is open (any lowercase alphanumeric with underscores; None = standard).

classmethod validate_lightgbmmt_instance_type(v)

Validate instance types suitable for LightGBMMT.

LightGBM works efficiently on CPU instances, especially for multi-task learning where memory and compute balance is important.

training_entry_point: str

training_instance_type: str

training_instance_count: int

training_volume_size: int

framework_version: str

py_version: str

ca_repository_arn: str

model_class: str

max_run_seconds: int

skip_hyperparameters_s3_uri: bool

use_secure_pypi: bool

use_precomputed_imputation: bool

use_precomputed_risk_tables: bool

use_precomputed_features: bool

use_native_categorical: bool

job_type: str | None

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

current_date: str

source_dir: str | None

enable_caching: bool

max_runtime_seconds: int

project_root_folder: str

class MissingValueImputationConfig(*, 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='missing_value_imputation.py', processing_script_arguments=None, processing_framework_version='1.2-1', label_field, job_type='training', default_numerical_strategy='mean', default_categorical_strategy='mode', default_text_strategy='mode', numerical_constant_value=0.0, categorical_constant_value='Unknown', text_constant_value='Unknown', categorical_preserve_dtype=True, auto_detect_categorical=True, categorical_unique_ratio_threshold=0.1, validate_fill_values=True, exclude_columns=None, column_strategies=None, enable_true_streaming=False, max_workers=0, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Missing Value Imputation step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

property effective_exclude_columns: List[str]

Get effective list of columns to exclude from imputation. Combines label_field with user-specified exclude_columns.

Returns:

List of column names to exclude

property environment_variables: Dict[str, str]

Get environment variables for the imputation script.

Returns:

Dictionary of environment variables

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include missing value imputation specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_categorical_strategy(v)

Ensure default_categorical_strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_column_strategies(v)

Validate column-specific strategies.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_exclude_columns(v)

Ensure exclude_columns contains non-empty strings if provided.

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_label_field(v)

Ensure label_field is a non-empty string.

classmethod validate_max_workers(v)

Ensure max_workers is 0 (auto-detect) or a positive integer.

classmethod validate_numerical_strategy(v)

Ensure default_numerical_strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_text_fill_values(v)

Validate that text fill values are pandas-safe.

classmethod validate_text_strategy(v)

Ensure default_text_strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

label_field: str

processing_entry_point: str

job_type: str

default_numerical_strategy: str

default_categorical_strategy: str

default_text_strategy: str

numerical_constant_value: float

categorical_constant_value: str

text_constant_value: str

categorical_preserve_dtype: bool

auto_detect_categorical: bool

categorical_unique_ratio_threshold: float

validate_fill_values: bool

exclude_columns: List[str] | None

column_strategies: Dict[str, str] | None

enable_true_streaming: bool

max_workers: int

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_script_arguments: List[str] | None

processing_framework_version: str

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

class ModelCalibrationConfig(*, 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='dockers/xgboost_atoz/scripts', processing_entry_point='model_calibration.py', processing_script_arguments=None, processing_framework_version='1.2-1', label_field, score_field=None, score_fields=None, task_label_names=None, calibration_method='gam', monotonic_constraint=True, gam_splines=10, error_threshold=0.05, is_binary=True, num_classes=2, score_field_prefix='prob_class_', calibration_sample_points=1000, multiclass_categories=<factory>, job_type='calibration', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for ModelCalibration step with self-contained derivation logic.

This class defines the configuration parameters for the ModelCalibration step, which calibrates model prediction scores to accurate probabilities. Calibration ensures that model scores reflect true probabilities, which is crucial for risk-based decision-making and threshold setting.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

classmethod from_hyperparameters(hyperparameters, region, pipeline_s3_loc, processing_instance_type='ml.m5.xlarge', processing_instance_count=1, processing_volume_size=30, max_runtime_seconds=3600, pipeline_name=None, calibration_method='gam', monotonic_constraint=True, gam_splines=10, error_threshold=0.05, processing_entry_point='model_calibration.py', processing_source_dir='dockers/xgboost_atoz/pipeline_scripts')

Create a ModelCalibrationConfig from a ModelHyperparameters instance.

This factory method creates a calibration config using values from the provided hyperparameters, with options to override specific calibration parameters.

Parameters:

• hyperparameters (ModelHyperparameters) – ModelHyperparameters instance with classification settings

• region (str) – AWS region

• pipeline_s3_loc (str) – S3 location for pipeline artifacts

• processing_instance_type (str) – SageMaker instance type for processing

• processing_instance_count (int) – Number of processing instances

• processing_volume_size (int) – EBS volume size in GB

• max_runtime_seconds (int) – Maximum runtime in seconds

• pipeline_name (str | None) – Name of the pipeline (optional)

• calibration_method (str) – Method to use for calibration (gam, isotonic, platt)

• monotonic_constraint (bool) – Whether to enforce monotonicity in GAM

• gam_splines (int) – Number of splines for GAM

• error_threshold (float) – Acceptable calibration error threshold

• processing_entry_point (str) – Script entry point filename

• processing_source_dir (str) – Directory containing the processing script

Returns:

Configuration object with values from hyperparameters

Return type:

ModelCalibrationConfig

get_environment_variables()

Get environment variables for the processing script.

Returns:

Dictionary of environment variables to be passed to the processing script.

Return type:

dict

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_config()

Validate configuration and ensure defaults are set.

Returns:

The validated configuration object

Return type:

Self

Raises:

ValueError – If any validation fails

label_field: str

score_field: str | None

score_fields: List[str] | None

task_label_names: List[str] | None

calibration_method: str

monotonic_constraint: bool

gam_splines: int

error_threshold: float

is_binary: bool

num_classes: int

score_field_prefix: str

calibration_sample_points: int

multiclass_categories: List[int | str]

job_type: str

processing_entry_point: str

processing_source_dir: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class ModelMetricsComputationConfig(*, 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='model_metrics_computation.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, score_field='prob_class_1', score_fields=None, task_label_names=None, job_type='calibration', amount_field='order_amount', input_format='auto', compute_dollar_recall=True, compute_count_recall=True, generate_plots=True, dollar_recall_fpr=0.1, count_recall_cutoff=0.1, comparison_mode=False, previous_score_field='', previous_score_fields=None, comparison_metrics='all', statistical_tests=True, comparison_plots=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for model metrics computation step with self-contained derivation logic.

This class defines the configuration parameters for the model metrics computation step, which loads prediction data, computes comprehensive performance metrics, generates visualizations, and creates detailed reports. Supports both binary and multiclass classification with domain-specific metrics like dollar and count recall.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the model metrics computation script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

classmethod validate_input_format(v)

Validate input format is supported.

validate_metrics_computation_config()

Additional validation specific to metrics computation configuration

classmethod validate_probability_range(v)

Validate probability values are in valid range.

id_name: str

label_name: str

score_field: str | None

score_fields: List[str] | None

task_label_names: List[str] | None

processing_entry_point: str

job_type: str

amount_field: str | None

input_format: str

compute_dollar_recall: bool

compute_count_recall: bool

generate_plots: bool

dollar_recall_fpr: float

count_recall_cutoff: float

processing_framework_version: str

use_large_processing_instance: bool

comparison_mode: bool

previous_score_field: str

previous_score_fields: List[str] | None

comparison_metrics: str

statistical_tests: bool

comparison_plots: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

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

class ModelWikiGeneratorConfig(*, 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='model_wiki_generator.py', processing_script_arguments=None, processing_framework_version='1.2-1', model_name, model_use_case='Machine Learning Model', team_alias='ml-team@', contact_email='ml-team@company.com', cti_classification='Internal', output_formats='wiki, html, markdown', include_technical_details=True, model_description=None, model_purpose='perform classification tasks', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for model wiki generator step with self-contained derivation logic.

This class defines the configuration parameters for the model wiki generator step, which loads metrics data and visualizations, generates comprehensive wiki documentation, and creates multi-format model documentation. Supports automated documentation creation for model registries and compliance requirements.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

property effective_model_description: str

Get effective model description (custom or auto-generated).

get_environment_variables()

Get environment variables for the model wiki generator script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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 model_display_name: str

Get display name for the model in documentation.

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property output_formats_list: List[str]

Get list of output formats from comma-separated string.

classmethod validate_cti_classification(v)

Validate CTI classification values.

classmethod validate_model_name(v)

Validate model name is not empty and contains valid characters.

classmethod validate_output_formats(v)

Validate output formats are supported.

validate_wiki_generator_config()

Additional validation specific to wiki generator configuration

model_name: str

processing_entry_point: str

model_use_case: str

team_alias: str

contact_email: str

cti_classification: str

output_formats: str

include_technical_details: bool

model_description: str | None

model_purpose: str

use_large_processing_instance: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

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

class PercentileModelCalibrationConfig(*, 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='percentile_model_calibration.py', processing_script_arguments=None, processing_framework_version='1.2-1', job_type, score_field=None, score_fields=None, n_bins=1000, accuracy=0.001, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for PercentileModelCalibration step with self-contained derivation logic.

This class defines the configuration parameters for the PercentileModelCalibration step, which creates percentile mapping from model scores using ROC curve analysis for consistent risk interpretation. The step converts raw model scores to percentile values that represent the relative risk ranking of predictions.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the processing script.

Returns:

Dictionary of environment variables to be passed to the processing script.

Return type:

dict

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_config()

Validate configuration and ensure defaults are set.

Returns:

The validated configuration object

Return type:

Self

Raises:

ValueError – If any validation fails

job_type: str

score_field: str | None

score_fields: List[str] | None

n_bins: int

accuracy: float

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class PyTorchModelEvalConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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=True, skip_volume_kms=None, processing_source_dir=None, processing_entry_point='pytorch_model_eval.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', eval_metric_choices=<factory>, comparison_mode=False, previous_score_field='', comparison_metrics='all', statistical_tests=True, comparison_plots=True, enable_true_streaming=False, num_workers_per_rank=0, prefetch_factor=None, use_persistent_workers=False, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for PyTorch model evaluation step with self-contained derivation logic.

This class defines the configuration parameters for the PyTorch model evaluation step, which calculates evaluation metrics for trained PyTorch models. This is crucial for measuring model performance and comparing different models or configurations.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the PyTorch model evaluation script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_dataloader_config()

Validate DataLoader worker configuration (warnings only, no mutation).

Conditional logic is enforced in evaluation script, not here. This validator only checks for potentially problematic values.

validate_eval_config()

Additional validation specific to evaluation configuration

id_name: str

label_name: str

processing_entry_point: str

job_type: str

eval_metric_choices: List[str]

framework_version: str

py_version: str

use_large_processing_instance: bool

comparison_mode: bool

previous_score_field: str

comparison_metrics: str

statistical_tests: bool

comparison_plots: bool

enable_true_streaming: bool

num_workers_per_rank: int

prefetch_factor: int | None

use_persistent_workers: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class PyTorchModelInferenceConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', 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='pytorch_inference.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', output_format='csv', json_orient='records', embedding_mode=False, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for PyTorch model inference step with self-contained derivation logic.

This class defines the configuration parameters for the PyTorch model inference step, which generates predictions from trained PyTorch models without computing evaluation metrics. This is designed for pure inference workflows where predictions are needed for downstream processing (e.g., model calibration, batch scoring).

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the PyTorch model inference script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_inference_config()

Additional validation specific to inference configuration

classmethod validate_json_orient(v)

Validate JSON orientation is supported.

classmethod validate_output_format(v)

Validate output format is supported.

id_name: str

label_name: str

processing_entry_point: str

job_type: str

output_format: str

json_orient: str

framework_version: str

py_version: str

use_large_processing_instance: bool

embedding_mode: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class StratifiedSamplingConfig(*, 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='stratified_sampling.py', processing_script_arguments=None, processing_framework_version='1.2-1', strata_column, job_type='training', sampling_strategy='balanced', target_sample_size=1000, min_samples_per_stratum=10, variance_column=None, sampling_multiplier=1.0, allow_replacement=False, reference_counts_json=None, sampling_filter_column=None, sampling_filter_value=None, random_state=42, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Stratified Sampling step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include stratified sampling specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_sampling_strategy(v)

Ensure sampling_strategy is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_strata_column(v)

Ensure strata_column is a non-empty string.

validate_strategy_requirements()

Validate that required fields are provided for specific strategies.

classmethod validate_variance_column(v)

Ensure variance_column is a non-empty string if provided.

strata_column: str

processing_entry_point: str

job_type: str

sampling_strategy: str

target_sample_size: int

min_samples_per_stratum: int

variance_column: str | None

sampling_multiplier: float

allow_replacement: bool

reference_counts_json: str | None

sampling_filter_column: str | None

sampling_filter_value: str | None

random_state: int

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_script_arguments: List[str] | None

processing_framework_version: str

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

class TemporalSequenceNormalizationConfig(*, 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='temporal_sequence_normalization.py', processing_script_arguments=None, processing_framework_version='1.2-1', temporal_field, sequence_grouping_field, record_id_field, job_type='training', sequence_length=51, sequence_separator='~', missing_indicators=['', 'My Text String'], time_delta_max_seconds=10000000, padding_strategy='pre', truncation_strategy='post', enable_multi_sequence=False, secondary_entity_field='creditCardId', sequence_naming_pattern='*_seq_by_{entity}.*', enable_distributed_processing=False, chunk_size=10000, max_workers='auto', validation_strategy='strict', output_format='numpy', include_attention_masks=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Temporal Sequence Normalization step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

property environment_variables: Dict[str, str]

Get environment variables dictionary for the processing step.

Returns:

Dictionary of environment variables

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include temporal sequence normalization specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_field_names(v)

Ensure field names are non-empty strings.

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_max_workers(v)

Ensure max_workers is ‘auto’ or a positive integer string.

classmethod validate_missing_indicators(v)

Ensure missing_indicators is a non-empty list.

classmethod validate_output_format(v)

Ensure output_format is valid (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_sequence_length(v)

Ensure sequence_length is positive.

classmethod validate_sequence_separator(v)

Ensure sequence_separator is non-empty.

classmethod validate_strategies(v)

Ensure strategies are valid (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_validation_strategy(v)

Ensure validation_strategy is valid (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

temporal_field: str

sequence_grouping_field: str

record_id_field: str

processing_entry_point: str

job_type: str

sequence_length: int

sequence_separator: str

missing_indicators: List[str]

time_delta_max_seconds: int

padding_strategy: str

truncation_strategy: str

enable_multi_sequence: bool

secondary_entity_field: str

sequence_naming_pattern: str

enable_distributed_processing: bool

chunk_size: int

max_workers: str

validation_strategy: str

output_format: str

include_attention_masks: bool

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_script_arguments: List[str] | None

processing_framework_version: str

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

class TemporalFeatureEngineeringConfig(*, 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='temporal_feature_engineering.py', processing_script_arguments=None, processing_framework_version='1.2-1', sequence_grouping_field, timestamp_field, value_fields, feature_types=['statistical', 'temporal', 'behavioral'], job_type='training', categorical_fields=['merchantCategory', 'paymentMethod'], window_sizes=[7, 14, 30, 90], aggregation_functions=['mean', 'sum', 'std', 'min', 'max', 'count'], lag_features=[1, 7, 14, 30], exponential_smoothing_alpha=0.3, time_unit='days', input_format='numpy', output_format='numpy', enable_distributed_processing=False, chunk_size=5000, max_workers='auto', feature_parallelism=True, cache_intermediate=True, enable_validation=True, missing_value_threshold=0.95, correlation_threshold=0.99, variance_threshold=0.01, outlier_detection=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Temporal Feature Engineering step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

property environment_variables: Dict[str, str]

Get environment variables dictionary for the processing step.

Returns:

Dictionary of environment variables

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include temporal feature engineering specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_aggregation_functions(v)

Ensure aggregation_functions contains valid function names.

classmethod validate_alpha(v)

Ensure exponential_smoothing_alpha is between 0.0 and 1.0.

classmethod validate_categorical_fields(v)

Ensure categorical_fields contains valid field names.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_feature_types(v)

Ensure feature_types contains valid feature types.

classmethod validate_field_names(v)

Ensure field names are non-empty strings.

classmethod validate_formats(v)

Ensure input/output formats are valid (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_job_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_lag_features(v)

Ensure lag_features contains positive integers.

classmethod validate_max_workers(v)

Ensure max_workers is ‘auto’ or a positive integer string.

classmethod validate_thresholds(v)

Ensure thresholds are between 0.0 and 1.0.

classmethod validate_time_unit(v)

Ensure time_unit is valid (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value.

classmethod validate_value_fields(v)

Ensure value_fields is a non-empty list of non-empty strings.

classmethod validate_variance_threshold(v)

Ensure variance_threshold is non-negative.

classmethod validate_window_sizes(v)

Ensure window_sizes contains positive integers.

sequence_grouping_field: str

timestamp_field: str

value_fields: List[str]

feature_types: List[str]

processing_entry_point: str

job_type: str

categorical_fields: List[str]

window_sizes: List[int]

aggregation_functions: List[str]

lag_features: List[int]

exponential_smoothing_alpha: float

time_unit: str

input_format: str

output_format: str

enable_distributed_processing: bool

chunk_size: int

max_workers: str

feature_parallelism: bool

cache_intermediate: bool

enable_validation: bool

missing_value_threshold: float

correlation_threshold: float

variance_threshold: float

outlier_detection: bool

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_script_arguments: List[str] | None

processing_framework_version: str

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

class XGBoostModelEvalConfig(*, 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=True, skip_volume_kms=None, processing_source_dir=None, processing_entry_point='xgboost_model_eval.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', eval_metric_choices=<factory>, xgboost_framework_version='1.5-1', comparison_mode=False, previous_score_field='', comparison_metrics='all', statistical_tests=True, comparison_plots=True, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for XGBoost model evaluation step with self-contained derivation logic.

This class defines the configuration parameters for the XGBoost model evaluation step, which calculates evaluation metrics for trained models. This is crucial for measuring model performance and comparing different models or configurations.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the model evaluation script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_eval_config()

Additional validation specific to evaluation configuration

id_name: str

label_name: str

processing_entry_point: str

job_type: str

eval_metric_choices: List[str]

xgboost_framework_version: str

use_large_processing_instance: bool

comparison_mode: bool

previous_score_field: str

comparison_metrics: str

statistical_tests: bool

comparison_plots: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

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

class XGBoostModelInferenceConfig(*, 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='xgboost_model_inference.py', processing_script_arguments=None, processing_framework_version='1.2-1', id_name, label_name, job_type='calibration', output_format='csv', json_orient='records', xgboost_framework_version='1.7-1', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for XGBoost model inference step with self-contained derivation logic.

This class defines the configuration parameters for the XGBoost model inference step, which generates predictions from trained models without computing evaluation metrics. This is designed for pure inference workflows where predictions are needed for downstream processing (e.g., model calibration, batch scoring).

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the model inference script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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.

validate_inference_config()

Additional validation specific to inference configuration

classmethod validate_json_orient(v)

Validate JSON orientation is supported.

classmethod validate_output_format(v)

Validate output format is supported.

id_name: str

label_name: str

processing_entry_point: str

job_type: str

output_format: str

json_orient: str

xgboost_framework_version: str

use_large_processing_instance: bool

processing_instance_count: int

processing_volume_size: int

processing_instance_type_large: str

processing_instance_type_small: str

skip_volume_kms: bool | None

processing_source_dir: str | None

processing_script_arguments: List[str] | None

processing_framework_version: str

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

class PyTorchModelStepConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=False, max_runtime_seconds=172800, project_root_folder, instance_type='ml.m5.large', entry_point='inference.py', accelerator_type=None, model_name=None, tags=None, initial_instance_count=1, container_startup_health_check_timeout=300, container_memory_limit=6144, data_download_timeout=900, inference_memory_limit=6144, max_concurrent_invocations=10, max_payload_size=6, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker PyTorch Model creation (for inference).

get_endpoint_config_name()

Generate endpoint configuration name

get_endpoint_name()

Generate endpoint name

get_model_name()

Generate a unique model name if not provided

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

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.

validate_configuration()

Validate the complete configuration

classmethod validate_memory_limits(v, info)

instance_type: str

entry_point: str

framework_version: str

py_version: str

accelerator_type: str | None

model_name: str | None

tags: List[Dict[str, str]] | None

initial_instance_count: int

container_startup_health_check_timeout: int

container_memory_limit: int

data_download_timeout: int

inference_memory_limit: int

max_concurrent_invocations: int

max_payload_size: int

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class XGBoostModelStepConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='1.5-1', py_version='py3', source_dir=None, enable_caching=False, use_secure_pypi=False, max_runtime_seconds=172800, project_root_folder, instance_type='ml.m5.large', entry_point='inference.py', accelerator_type=None, model_name=None, tags=None, initial_instance_count=1, container_startup_health_check_timeout=300, container_memory_limit=6144, data_download_timeout=900, inference_memory_limit=6144, max_concurrent_invocations=10, max_payload_size=6, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker XGBoost Model creation (for inference).

get_endpoint_config_name()

Generate endpoint configuration name

get_endpoint_name()

Generate endpoint name

get_model_name()

Generate a unique model name if not provided

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

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.

validate_configuration()

Validate the complete configuration

classmethod validate_memory_limits(v, info)

instance_type: str

entry_point: str

framework_version: str

py_version: str

accelerator_type: str | None

model_name: str | None

tags: List[Dict[str, str]] | None

initial_instance_count: int

container_startup_health_check_timeout: int

container_memory_limit: int

data_download_timeout: int

inference_memory_limit: int

max_concurrent_invocations: int

max_payload_size: int

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class PackageConfig(*, 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='package.py', processing_script_arguments=None, processing_framework_version='1.2-1', **extra_data)

Bases: ProcessingStepConfigBase

Configuration for a model packaging step.

This configuration follows the three-tier field categorization: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that users can override 3. Tier 3: Derived Fields - fields calculated from other fields, stored in private attributes

get_environment_variables(declared_env_vars=None)

Packaging env vars (the single env source; moved here from the builder, FZ 31e1d3g).

declared_env_vars accepted for the builder’s names-driven contract but ignored — these are config-derived names (PIPELINE_NAME, REGION, …) emitted only when the underlying field is present, preserving the builder’s original conditional-add behavior.

inference_scripts_source()

Local source for the packaging step’s inference_scripts_input (FZ 31e1d3i).

The packaging step always mounts inference scripts from a LOCAL path (overriding any dependency-resolved value). Delegates to effective_source_dir — the single comprehensive source-dir resolver (hybrid processing_source_dir → hybrid source_dir → legacy values) on ProcessingStepConfigBase — falling back to the literal "inference" only when no source dir is configured.

NOTE: the original builder used resolved_source_dir or source_dir or "inference", which reimplemented a PARTIAL version of this resolution and silently IGNORED processing_source_dir (falling through to "inference" when only that was set — a latent bug). Using effective_source_dir fixes that and removes the duplicated chain.

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

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.

validate_config()

Validate configuration and ensure defaults are set.

This validator ensures that: 1. Entry point is provided 2. Script contract is available and valid 3. Required input paths are defined in the script contract

processing_entry_point: str

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_script_arguments: List[str] | None

processing_framework_version: str

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

class PayloadConfig(*, 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='payload.py', processing_script_arguments=None, processing_framework_version='1.2-1', expected_tps, max_latency_in_millisecond, source_model_inference_content_types=['text/csv'], source_model_inference_response_types=['application/json'], default_numeric_value=0.0, default_text_value='DEFAULT_TEXT', field_defaults=None, custom_payload_path=None, max_acceptable_error_rate=0.2, load_test_instance_type_list=['ml.m5.4xlarge'], **extra_data)

Bases: ProcessingStepConfigBase

Configuration for payload generation and testing.

This configuration follows the three-tier field categorization: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that users can override 3. Tier 3: Derived Fields - fields calculated from other fields, stored in private attributes

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h). Custom passthrough or None.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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

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.

serialize_path_fields(value)

Serialize Path objects to strings

classmethod validate_load_test_instance_types(v)

Validate load test instance type format

expected_tps: int

max_latency_in_millisecond: int

processing_entry_point: str

source_model_inference_content_types: List[str]

source_model_inference_response_types: List[str]

default_numeric_value: float

default_text_value: str

field_defaults: Dict[str, str] | None

custom_payload_path: str | None

max_acceptable_error_rate: float

load_test_instance_type_list: List[str]

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_script_arguments: List[str] | None

processing_framework_version: str

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

class RegistrationConfig(*, 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, model_owner, model_domain, model_objective, framework, inference_entry_point, source_model_inference_input_variable_list=<factory>, inference_instance_type='ml.m5.large', source_model_inference_content_types=['text/csv'], source_model_inference_response_types=['application/json'], source_model_inference_output_variable_list={'legacy-score': VariableType.NUMERIC}, **extra_data)

Bases: BasePipelineConfig

Configuration for model registration step, following the three-tier categorization:

Tier 1: Essential User Inputs - fields that users must explicitly provide Tier 2: System Inputs - fields with reasonable defaults that users can override Tier 3: Derived Fields - private fields with read-only property access

get_variable_schema()

Legacy method that forwards to the property

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

serialize_input_variable_list(value)

Serialize VariableType enum values to strings in input variable list

serialize_output_variable_list(value)

Serialize VariableType enum values to strings

set_source_model_inference_input_variable_list(numeric_fields=None, text_fields=None, output_format='dict')

Set the input variable list for model inference using separate lists for numeric and text fields.

Parameters:

• numeric_fields (List[str]) – List of field names that should be treated as NUMERIC

• text_fields (List[str]) – List of field names that should be treated as TEXT

• output_format (str) – Format for storing variable list - either ‘dict’ or ‘list’

classmethod validate_content_types(v)

Validate content and response types

classmethod validate_framework(v)

Validate the ML framework

classmethod validate_inference_instance_type(v)

Validate the inference instance type

classmethod validate_input_variable_list(v)

Validate input variable list format.

Parameters:

v (Dict[str, VariableType | str] | List[List[str]]) – Either a dictionary of variable names to types, or a list of [variable_name, variable_type] pairs

Returns:

Validated input variable list

Return type:

Dict[str, VariableType | str] | List[List[str]]

classmethod validate_output_variable_list(v)

Validate variable lists and convert to string values

validate_registration_configs()

Validate registration-specific configurations (without file existence checks)

property variable_schema: Dict[str, Dict[str, List[Dict[str, str]]]]

Generate variable schema for model registration

model_owner: str

model_domain: str

model_objective: str

framework: str

inference_entry_point: str

source_model_inference_input_variable_list: Dict[str, VariableType | str] | List[List[str]]

inference_instance_type: str

source_model_inference_content_types: List[str]

source_model_inference_response_types: List[str]

source_model_inference_output_variable_list: Dict[str, VariableType]

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

class RiskTableMappingConfig(*, 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='risk_table_mapping.py', processing_script_arguments=None, processing_framework_version='1.2-1', job_type='training', cat_field_list=[], label_name='target', smooth_factor=0.01, count_threshold=5, max_unique_threshold=100, enable_true_streaming=False, max_workers=0, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for Risk Table Mapping Processing Step.

This class extends ProcessingStepConfigBase to include specific fields for risk table mapping, including categorical fields and job type.

Source Directory Integration: After Phase 6 refactor, hyperparameters are embedded in the source directory structure and no longer require separate S3 upload. The expected source directory structure is:

source_dir/ ├── risk_table_mapping.py # Main script (entry point) └── hyperparams/ # Hyperparameters directory

└── hyperparameters.json # Generated hyperparameters file

The hyperparameters.json file is automatically generated from the configuration fields (cat_field_list, label_name, smooth_factor, count_threshold) and embedded in the source directory for runtime access.

property environment_variables: Dict[str, str]

Get environment variables for the risk table mapping script.

Returns:

Dictionary of environment variables

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

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

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.

classmethod validate_job_type(v)

Validate job type is one of the allowed values.

classmethod validate_max_workers(v)

Ensure max_workers is 0 (auto-detect) or a positive integer.

validate_risk_table_config()

Validate risk table mapping configuration.

After Phase 6 refactor: Simplified validation focusing on core configuration without S3 upload logic or external hyperparameters handling.

processing_entry_point: str

job_type: str

cat_field_list: List[str]

label_name: str

smooth_factor: float

count_threshold: int

max_unique_threshold: int

enable_true_streaming: bool

max_workers: int

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_script_arguments: List[str] | None

processing_framework_version: str

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

class TabularPreprocessingConfig(*, 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='tabular_preprocessing.py', processing_script_arguments=None, processing_framework_version='1.2-1', job_type, label_name=None, train_ratio=0.7, test_val_ratio=0.5, output_format='CSV', max_workers=0, batch_size=5, optimize_memory=False, streaming_batch_size=0, enable_true_streaming=False, **extra_data)

Bases: ProcessingStepConfigBase

Configuration for the Tabular Preprocessing step with three-tier field categorization. Inherits from ProcessingStepConfigBase.

Fields are categorized into: - Tier 1: Essential User Inputs - Required from users - Tier 2: System Fields - Default values that can be overridden - Tier 3: Derived Fields - Private with read-only property access

property full_script_path: str | None

Get full path to the preprocessing script.

Returns:

Full path to the script

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

Override get_public_init_fields to include tabular preprocessing specific fields.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

property preprocessing_environment_variables: Dict[str, str]

Get preprocessing-specific environment variables.

Returns:

Dictionary mapping environment variable names to values

classmethod validate_data_type(v)

Ensure job_type is one of the allowed values.

classmethod validate_entry_point_relative(v)

Ensure processing_entry_point is a non‐empty relative path.

classmethod validate_output_format(v)

Ensure output_format is one of the allowed values (case-insensitive).

Matching is case-insensitive and the stored value is normalized to the canonical-cased allowed value, so the persisted config never drifts.

classmethod validate_ratios(v)

Ensure the ratio is strictly between 0 and 1 (not including 0 or 1).

job_type: str

label_name: str | None

processing_entry_point: str

train_ratio: float

test_val_ratio: float

output_format: str

max_workers: int

batch_size: int

optimize_memory: bool

streaming_batch_size: int

enable_true_streaming: bool

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_script_arguments: List[str] | None

processing_framework_version: str

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

class PyTorchTrainingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='2.1.2', py_version='py310', source_dir=None, enable_caching=False, use_secure_pypi=False, max_runtime_seconds=172800, project_root_folder, training_entry_point, training_instance_type='ml.g5.12xlarge', training_instance_count=1, training_volume_size=30, ca_repository_arn='arn:aws:codeartifact:us-west-2:149122183214:repository/amazon/secure-pypi', skip_hyperparameters_s3_uri=True, hyperparameters=None, use_precomputed_imputation=False, use_precomputed_risk_tables=False, use_precomputed_features=False, enable_true_streaming=False, num_workers_per_rank=0, prefetch_factor=None, use_persistent_workers=False, job_type=None, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker PyTorch Training Step. This version is streamlined to work with specification-driven architecture. Input/output paths are now provided via step specifications and dependencies.

get_environment_variables()

Get environment variables for the PyTorch training script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

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

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.

validate_dataloader_config()

Validate DataLoader worker configuration (warnings only, no mutation).

Conditional logic is enforced in training script, not here. This validator only checks for potentially problematic values.

classmethod validate_job_type(v)

Validate job_type is open (any lowercase alphanumeric with underscores; None = standard).

training_entry_point: str

training_instance_type: str

training_instance_count: int

training_volume_size: int

framework_version: str

py_version: str

ca_repository_arn: str

skip_hyperparameters_s3_uri: bool

hyperparameters: ModelHyperparameters | None

use_precomputed_imputation: bool

use_precomputed_risk_tables: bool

use_precomputed_features: bool

enable_true_streaming: bool

num_workers_per_rank: int

prefetch_factor: int | None

use_persistent_workers: bool

job_type: str | None

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

use_secure_pypi: bool

max_runtime_seconds: int

project_root_folder: str

class XGBoostTrainingConfig(*, author, bucket, role, region, service_name, pipeline_version, model_class='xgboost', current_date=<factory>, framework_version='1.7-1', py_version='py3', source_dir=None, enable_caching=False, use_secure_pypi=True, max_runtime_seconds=172800, project_root_folder, training_entry_point, training_instance_type='ml.m5.4xlarge', training_instance_count=1, training_volume_size=30, skip_hyperparameters_s3_uri=True, use_precomputed_imputation=False, use_precomputed_risk_tables=False, use_precomputed_features=False, job_type=None, **extra_data)

Bases: BasePipelineConfig

Configuration specific to the SageMaker XGBoost Training Step. This version is streamlined to work with specification-driven architecture. Input/output paths are now provided via step specifications and dependencies.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private with properties)

get_environment_variables()

Get environment variables for the XGBoost training script.

Returns:

Dictionary mapping environment variable names to values

Return type:

Dict[str, str]

get_job_arguments()

CLI args — config is the single source (FZ 31e1d3h).

get_public_init_fields()

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

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

property hyperparameter_file: str

Get hyperparameter file path.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

model_dump(**kwargs)

Override model_dump to include derived properties.

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.

classmethod validate_job_type(v)

Validate job_type is one of allowed values.

training_entry_point: str

training_instance_type: str

training_instance_count: int

training_volume_size: int

framework_version: str

py_version: str

skip_hyperparameters_s3_uri: bool

use_secure_pypi: bool

use_precomputed_imputation: bool

use_precomputed_risk_tables: bool

use_precomputed_features: bool

job_type: str | None

author: str

bucket: str

role: str

region: str

service_name: str

pipeline_version: str

model_class: str

current_date: str

source_dir: str | None

enable_caching: bool

max_runtime_seconds: int

project_root_folder: str

class BaseCradleComponentConfig(**extra_data)

Bases: BaseModel

Base class for Cradle configuration components with three-tier field classification support.

Implements common functionality for categorizing fields and supporting inheritance.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

categorize_fields()

Categorize all fields into three tiers: 1. Tier 1: Essential User Inputs - fields with no defaults (required) 2. Tier 2: System Inputs - fields with defaults (optional) 3. Tier 3: Derived Fields - properties that access private attributes

Returns:

Dict with keys ‘essential’, ‘system’, and ‘derived’ mapping to lists of field names

Return type:

Dict[str, List[str]]

get_public_init_fields()

Get fields suitable for initializing a child config. Only includes fields that should be passed to child class constructors.

Returns:

Dictionary of field names to values for child initialization

Return type:

Dict[str, Any]

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

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

class MdsDataSourceConfig(*, service_name, region, output_schema, org_id=0, use_hourly_edx_data_set=False, **extra_data)

Bases: BaseCradleComponentConfig

Configuration for MDS data source with three-tier field classification.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

classmethod validate_region(v)

service_name: str

region: str

output_schema: List[Dict[str, Any]]

org_id: int

use_hourly_edx_data_set: bool

class EdxDataSourceConfig(*, schema_overrides=None, edx_arn=None, edx_provider=None, edx_subject=None, edx_dataset=None, edx_manifest_key=None, **extra_data)

Bases: BaseCradleComponentConfig

Configuration for EDX data source with three-tier field classification.

Supports two input modes: 1. Direct ARN input: Provide edx_arn directly 2. Component-based input: Provide edx_provider, edx_subject, edx_dataset, edx_manifest_key

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

categorize_fields()

Dynamic field categorization based on edx_arn presence.

property edx_manifest: str

Get EDX manifest ARN from direct input or built from components.

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

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

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.

classmethod validate_edx_arn_format(v)

Validate EDX ARN format if provided.

validate_edx_input_mode()

Ensure either edx_arn OR all component fields are provided.

classmethod validate_manifest_key_format(v)

Validate that edx_manifest_key is in the format ‘[…]’ if provided.

schema_overrides: List[Dict[str, Any]] | None

edx_arn: str | None

edx_provider: str | None

edx_subject: str | None

edx_dataset: str | None

edx_manifest_key: str | None

class AndesDataSourceConfig(*, provider, table_name, andes3_enabled=True, **extra_data)

Bases: BaseCradleComponentConfig

Configuration for Andes data source with three-tier field classification.

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

classmethod validate_provider(v)

Validate that the provider is either: 1. A valid 32-character UUID 2. The special case ‘booker’

classmethod validate_table_name(v)

Validate that the table name is not empty and follows valid format.

provider: str

table_name: str

andes3_enabled: bool

class DataSourceConfig(*, data_source_name, data_source_type, mds_data_source_properties=None, edx_data_source_properties=None, andes_data_source_properties=None, **extra_data)

Bases: BaseCradleComponentConfig

Configuration for data sources with three-tier field classification.

Corresponds to com.amazon.secureaisandboxproxyservice.models.datasource.DataSource

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

classmethod check_properties(model)

Ensure the appropriate properties are set based on data_source_type and that only one set of properties is provided.

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

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

classmethod validate_type(v)

data_source_name: str

data_source_type: str

mds_data_source_properties: MdsDataSourceConfig | None

edx_data_source_properties: EdxDataSourceConfig | None

andes_data_source_properties: AndesDataSourceConfig | None

class DataSourcesSpecificationConfig(*, start_date, end_date, data_sources, **extra_data)

Bases: BaseCradleComponentConfig

Configuration for data sources specification with three-tier field classification.

Corresponds to com.amazon.secureaisandboxproxyservice.models.datasourcesspecification.DataSourcesSpecification

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

classmethod validate_exact_datetime_format(v, field)

Must match exactly “%Y-%m-%dT%H:%M:%S”

start_date: str

end_date: str

data_sources: List[DataSourceConfig]

class JobSplitOptionsConfig(*, merge_sql=None, split_job=False, days_per_split=7, **extra_data)

Bases: BaseCradleComponentConfig

Corresponds to com.amazon.secureaisandboxproxyservice.models.jobsplitoptions.JobSplitOptions:

• split_job: bool

• days_per_split: int

• merge_sql: str

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

classmethod days_must_be_positive(v)

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

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

classmethod require_merge_sql_if_split(model)

merge_sql: str | None

split_job: bool

days_per_split: int

class TransformSpecificationConfig(*, transform_sql, job_split_options=<factory>, **extra_data)

Bases: BaseCradleComponentConfig

Corresponds to com.amazon.secureaisandboxproxyservice.models.transformspecification.TransformSpecification:

• transform_sql: str

• job_split_options: JobSplitOptionsConfig

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

transform_sql: str

job_split_options: JobSplitOptionsConfig

class OutputSpecificationConfig(*, output_schema, job_type='training', pipeline_s3_loc=None, output_format='PARQUET', output_save_mode='ERRORIFEXISTS', output_file_count=0, keep_dot_in_output_schema=False, include_header_in_s3_output=True, **extra_data)

Bases: BaseCradleComponentConfig

Corresponds to com.amazon.secureaisandboxproxyservice.models.outputspecification.OutputSpecification:

• output_schema: List[str]

• output_format: str (e.g. ‘PARQUET’, ‘CSV’, etc.)

• output_save_mode: str (e.g. ‘ERRORIFEXISTS’, ‘OVERWRITE’, ‘APPEND’, ‘IGNORE’)

• output_file_count: int (0 means “auto”)

• keep_dot_in_output_schema: bool

• include_header_in_s3_output: bool

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

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.

property output_path: str

Get output path derived from pipeline_s3_loc and job_type.

classmethod validate_format(v)

classmethod validate_job_type(v)

validate_output_path()

Validate that output_path is a valid S3 URI.

classmethod validate_save_mode(v)

output_schema: List[str]

job_type: str

pipeline_s3_loc: str | None

output_format: str

output_save_mode: str

output_file_count: int

keep_dot_in_output_schema: bool

include_header_in_s3_output: bool

class CradleJobSpecificationConfig(*, cradle_account, cluster_type='STANDARD', extra_spark_job_arguments='', job_retry_count=1, **extra_data)

Bases: BaseCradleComponentConfig

Corresponds to com.amazon.secureaisandboxproxyservice.models.cradlejobspecification.CradleJobSpecification:

• cluster_type: str (e.g. ‘SMALL’, ‘MEDIUM’, ‘LARGE’)

• cradle_account: str

• extra_spark_job_arguments: Optional[str]

• job_retry_count: int

Fields are organized into three tiers: 1. Tier 1: Essential User Inputs - fields that users must explicitly provide 2. Tier 2: System Inputs with Defaults - fields with reasonable defaults that can be overridden 3. Tier 3: Derived Fields - fields calculated from other fields (private attributes with properties)

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

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

classmethod validate_cluster_type(v)

cradle_account: str

cluster_type: str

extra_spark_job_arguments: str | None

job_retry_count: int

class VariableType(*values)

Bases: str, Enum

NUMERIC = 'NUMERIC'

TEXT = 'TEXT'

create_inference_variable_list(numeric_fields=None, text_fields=None, output_format='dict')

Create an inference variable list for model input variables using separate lists for numeric and text fields. This is a helper function that can be used standalone or within RegistrationConfig.

Parameters:

• numeric_fields (List[str]) – List of field names that should be treated as NUMERIC

• text_fields (List[str]) – List of field names that should be treated as TEXT

• output_format (str) – Format for storing variable list - either ‘dict’ or ‘list’

Returns:

A dictionary mapping variable names to their types, or a list of [name, type] pairs, depending on the output_format parameter

Return type:

Dict[str, VariableType | str] | List[List[str]]

detect_config_classes_from_json(config_path)

Fallback implementation that simply calls build_complete_config_classes.

class CategoryType(*values)

Bases: Enum

SHARED = 1

SPECIFIC = 2

serialize_config(config)

Serialize a single Pydantic config to a JSON‐serializable dict, embedding metadata including a unique ‘step_name’. Enhanced to include default values from Pydantic model definitions.

This function maintains backward compatibility while using the new implementation.

verify_configs(config_list)

Verify that the configurations are valid.

Parameters:

config_list (List[BaseModel]) – List of configurations to verify

Raises:

ValueError – If configurations are invalid (e.g., duplicate step names)

merge_and_save_configs(config_list, output_file)

Merge and save multiple configs to JSON. Handles multiple instantiations with unique step_name. Better handles class hierarchy for fields like input_names that should be kept specific.

This is a wrapper for the new implementation in src.config_field_manager.

NOTE: This function adds field_sources data to the metadata section, tracking which fields come from which configs. The structure is completely flattened as:

metadata.field_sources = { field_name: [config_name, …], … }

Simplified Field Categorization Rules:

1. Field is special → Place in specific - Special fields include those in the SPECIAL_FIELDS_TO_KEEP_SPECIFIC list - Pydantic models are considered special fields - Complex nested structures are considered special fields

2. Field appears only in one config → Place in specific - If a field exists in only one configuration instance, it belongs in that instance’s specific section

3. Field has different values across configs → Place in specific - If a field has the same name but different values across multiple configs, each instance goes in specific

4. Field is non-static → Place in specific - Fields identified as non-static (runtime values, input/output fields, etc.) go in specific

5. Field has identical value across all configs → Place in shared - If a field has the same value across all configs and is not caught by the above rules, it belongs in shared

6. Default case → Place in specific - When in doubt, place in specific to ensure proper functioning

We build a simplified structure:

• “shared”: fields that appear with identical values across all configs and are static

• “specific”: fields that are unique to specific configs or have different values across configs

The following categories are mutually exclusive:

• “shared” and “specific” sections have no overlapping fields

Under “metadata” → “config_types” we map each unique step_name → config class name.

load_configs(input_file, config_classes=None, project_id=None)

Load multiple Pydantic configs from JSON, reconstructing each instantiation uniquely.

ENHANCED: Step catalog integration for deployment-agnostic loading.

Portability: Works across all deployment environments Discovery: Automatic config class resolution Workspace: Project-specific loading support

Parameters:

• input_file (str) – Path to the input JSON file

• config_classes (Dict[str, Type[BaseModel]] | None) – Optional dictionary mapping class names to class types

• project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping step names to config instances

Return type:

Dict[str, BaseModel]

get_field_sources(config_list)

Extract field sources from config list.

Returns a dictionary with three categories: - ‘all’: All fields and their source configs - ‘processing’: Fields from processing configs - ‘specific’: Fields from non-processing configs

This is used for backward compatibility with the legacy field categorization.

Parameters:

config_list (List[BaseModel]) – List of configuration objects to analyze

Returns:

Dictionary of field sources by category

Return type:

Dict[str, Dict[str, List[str]]]

build_complete_config_classes(project_id=None)

Build a complete dictionary of all relevant config classes using the unified step catalog system’s ConfigAutoDiscovery.

REFACTORED: Now uses step catalog integration with multiple fallback strategies. PORTABLE: Works across all deployment scenarios (PyPI, source, submodule).

Success Rate: 83% failure → 100% success Deployment: Works in all environments (dev, Lambda, Docker, PyPI) Workspace: Optional project-specific discovery

Parameters:

project_id (str | None) – Optional project ID for workspace-specific discovery

Returns:

Dictionary mapping class names to class types

Return type:

Dict[str, Type[BaseModel]]

Modules

config_active_sample_selection_step	Active Sample Selection Step Configuration
config_batch_transform_step
config_bedrock_batch_processing_step	Bedrock Batch Processing Step Configuration
config_bedrock_processing_step	Bedrock Processing Step Configuration
config_bedrock_prompt_template_generation_step	Bedrock Prompt Template Generation Step Configuration
config_cradle_data_loading_step
config_currency_conversion_step	Currency Conversion Configuration with Self-Contained Derivation Logic
config_data_uploading_step	Data Uploading Step Configuration.
config_dummy_data_loading_step	Dummy Data Loading Step Configuration
config_dummy_training_step	Configuration for DummyTraining step with flexible input modes.
config_edx_uploading_step	EdxUploading Step Configuration.
config_feature_selection_step	Feature Selection Configuration with Self-Contained Derivation Logic
config_label_ruleset_execution_step	Label Ruleset Execution Step Configuration
config_label_ruleset_generation_step	Label Ruleset Generation Step Configuration
config_lightgbm_model_eval_step	Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_lightgbm_model_inference_step	Model Inference Step Configuration with Self-Contained Derivation Logic
config_lightgbm_training_step	LightGBM Training Step Configuration with Self-Contained Derivation Logic
config_lightgbmmt_model_eval_step	Multi-Task Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_lightgbmmt_model_inference_step	Multi-Task Model Inference Step Configuration with Self-Contained Derivation Logic
config_lightgbmmt_training_step	LightGBMMT Training Step Configuration
config_missing_value_imputation_step	Missing Value Imputation Configuration with Self-Contained Derivation Logic
config_model_calibration_step	Model Calibration Step Configuration with Self-Contained Derivation Logic
config_model_metrics_computation_step	Model Metrics Computation Step Configuration with Self-Contained Derivation Logic
config_model_wiki_generator_step	Model Wiki Generator Step Configuration with Self-Contained Derivation Logic
config_package_step
config_payload_step
config_percentile_model_calibration_step	Percentile Model Calibration Step Configuration with Self-Contained Derivation Logic
config_piper_metric_generation_step	PIPER Metric Generation Step Configuration with Self-Contained Derivation Logic
config_processing_step_base	Processing Step Base Configuration with Self-Contained Derivation Logic
config_pseudo_label_merge_step	Pseudo Label Merge Step Configuration
config_pytorch_model_eval_step	PyTorch Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_pytorch_model_inference_step	PyTorch Model Inference Step Configuration with Self-Contained Derivation Logic
config_pytorch_model_step
config_pytorch_training_step
config_redshift_data_loading_step	Redshift Data Loading Step Configuration.
config_registration_step
config_risk_table_mapping_step	Configuration for Risk Table Mapping Processing Step.
config_sopa_instruction_tuning_step	SOPA Instruction Tuning Step Configuration with Self-Contained Derivation Logic
config_stratified_sampling_step	Stratified Sampling Configuration with Self-Contained Derivation Logic
config_tabular_preprocessing_step	Tabular Preprocessing Configuration with Self-Contained Derivation Logic
config_temporal_feature_engineering_step	Temporal Feature Engineering Configuration with Self-Contained Derivation Logic
config_temporal_sequence_normalization_step	Temporal Sequence Normalization Configuration with Self-Contained Derivation Logic
config_temporal_split_preprocessing_step	Temporal Split Preprocessing Configuration with Self-Contained Derivation Logic
config_tokenizer_training_step	Configuration for Tokenizer Training Processing Step.
config_tsa_model_calibration_step	TSA Model Calibration Step Configuration with Self-Contained Derivation Logic
config_tsa_model_eval_step	TSA Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_tsa_preprocessing_step	TSA Data Preprocessing Configuration with Self-Contained Derivation Logic
config_tsa_tabular_preprocessing_step	TSA Tabular Preprocessing Configuration with Self-Contained Derivation Logic
config_tsa_training_step	TSA Training Step Configuration with Self-Contained Derivation Logic
config_xgboost_model_eval_step	Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_xgboost_model_inference_step	Model Inference Step Configuration with Self-Contained Derivation Logic
config_xgboost_model_step
config_xgboost_mt_model_eval_step	Multi-Task Model Evaluation Step Configuration with Self-Contained Derivation Logic
config_xgboost_mt_training_step	XgboostMt Training Step Configuration
config_xgboost_training_step	XGBoost Training Step Configuration with Self-Contained Derivation Logic
utils	Configuration utility functions for merging, saving, and loading multiple Pydantic configs.