cursus.steps.configs.config_label_ruleset_generation_step

Label Ruleset Generation Step Configuration

This module implements the configuration class for the Label Ruleset Generation step using the three-tier design pattern for optimal user experience and maintainability.

class ComparisonOperator(*values)

Bases: str, Enum

Supported comparison operators for rule conditions.

Categories: - Comparison: equals, not_equals, gt, gte, lt, lte - Collection: in_collection, not_in_collection - String: contains, not_contains, starts_with, ends_with, regex_match - Null: is_null, is_not_null

EQUALS = 'equals'

NOT_EQUALS = 'not_equals'

GT = '>'

GTE = '>='

LT = '<'

LTE = '<='

IN = 'in'

NOT_IN = 'not_in'

CONTAINS = 'contains'

NOT_CONTAINS = 'not_contains'

STARTS_WITH = 'starts_with'

ENDS_WITH = 'ends_with'

REGEX_MATCH = 'regex_match'

IS_NULL = 'is_null'

IS_NOT_NULL = 'is_not_null'

class RuleCondition(*, field=None, operator=None, value=None, all_of=None, any_of=None, none_of=None)

Bases: BaseModel

Single condition in a rule.

Supports nested logical operators (all_of, any_of, none_of) and leaf conditions with field comparisons using validated operators.

All fields are optional (Tier 2) but mutually exclusive validation ensures conditions are either leaf (field/operator/value) or logical (all_of/any_of/none_of).

field: str | None

operator: ComparisonOperator | None

value: Any | None

all_of: List[RuleCondition] | None

any_of: List[RuleCondition] | None

none_of: List[RuleCondition] | None

validate_condition_structure()

Validate that condition is either a leaf or a logical operator.

to_script_format()

Convert to format expected by script.

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

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

class LabelConfig(*, output_label_name, output_label_type, label_values, label_mapping, default_label, evaluation_mode='priority', sparse_representation=True)

Bases: BaseModel

Pydantic model for label configuration with multi-label support.

Supports three modes via output_label_type: - ‘binary’: Single binary column - ‘multiclass’: Single multiclass column - ‘multilabel’: Multiple columns (new)

Follows three-tier design: - Tier 1: Required user inputs - Tier 2: Optional user inputs with defaults

output_label_name: str | List[str]

output_label_type: str

label_values: List[int | str] | Dict[str, List[int | str]]

label_mapping: Dict[str, str] | Dict[str, Dict[str, str]]

default_label: int | str | Dict[str, int | str]

evaluation_mode: str

sparse_representation: bool

classmethod validate_label_type(v)

Validate label_type is valid.

classmethod validate_evaluation_mode(v)

Validate evaluation mode.

validate_consistency()

Validate fields match output_label_type.

validate_default_label()

Validate default_label is in label_values.

validate_binary_constraints()

Validate binary classification uses [0, 1] values.

to_script_format()

Convert to format expected by script.

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

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

class FieldConfig(*, required_fields, field_types, optional_fields=<factory>)

Bases: BaseModel

Pydantic model for field configuration.

Defines the schema of fields that can be referenced in rules.

Follows three-tier design: - Tier 1: Required user inputs - Tier 2: Optional user inputs with defaults

required_fields: List[str]

field_types: Dict[str, str]

optional_fields: List[str]

classmethod validate_field_types(v)

Validate field types are valid.

validate_all_fields_have_types()

Validate all declared fields have types.

to_script_format()

Convert to format expected by script.

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

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

class RuleDefinition(*, name, priority, conditions, output_label, enabled=True, description='')

Bases: BaseModel

Pydantic model for a single rule definition.

Defines a classification rule with conditions and output label. The rule_id is auto-generated and should not be provided by users.

Follows three-tier design: - Tier 1: Required user inputs - Tier 2: Optional user inputs with defaults - Tier 3: Derived fields (private, auto-generated)

name: str

priority: int

conditions: RuleCondition

output_label: int | str | Dict[str, int | str]

enabled: bool

description: str

property rule_id: str

Get auto-generated unique rule identifier.

classmethod validate_name(v)

Validate name is not empty.

validate_output_label()

Validate output_label format.

to_script_format()

Convert to format expected by script.

model_dump(**kwargs)

Override model_dump to include auto-generated rule_id.

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', '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.

class RulesetDefinitionList(*, rules)

Bases: BaseModel

Pydantic model for a list of rule definitions with validation.

Ensures rule IDs are unique and provides utility methods.

rules: List[RuleDefinition]

classmethod validate_unique_rule_ids(v)

Validate all rule IDs are unique.

to_script_format()

Convert all rules to format expected by script.

to_json(**kwargs)

Convert to JSON string in script format.

get_rule_ids()

Get list of all rule IDs.

get_rule_by_id(rule_id)

Get rule by ID.

sort_by_priority()

Return new RulesetDefinitionList sorted by priority.

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

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

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)

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

property environment_variables: Dict[str, str]

Get environment variables for the processing step.

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

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

model_dump(**kwargs)

Override model_dump to include derived properties.

initialize_derived_fields()

Initialize all derived fields once after validation.

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

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]

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.

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

load_rules_from_json(json_data)

Load rules from JSON string with validation.

Parameters:

json_data (str) – JSON string containing rule definitions

Returns:

Validated RulesetDefinitionList

Raises:

• ValueError – If JSON is invalid or rules don’t validate

• pydantic.ValidationError – If rule data doesn’t match schema

Return type:

RulesetDefinitionList

load_rules_from_dict(data)

Load rules from dictionary/list data with validation.

Parameters:

data (Any) – Dictionary or list containing rule definitions

Returns:

Validated RulesetDefinitionList

Raises:

pydantic.ValidationError – If rule data doesn’t match schema

Return type:

RulesetDefinitionList