cursus.api.factory

DAG Configuration Factory API

This module provides interactive pipeline configuration generation through a step-by-step workflow. The system leverages existing Pydantic field definitions and the three-tier configuration system to create user-friendly configuration interfaces for pipeline development.

Key Components: - DAGConfigFactory: Main interactive factory for step-by-step configuration generation - ConfigClassMapper: Maps DAG nodes to configuration classes using registry system - ConfigurationGenerator: Generates final configuration instances with base config inheritance - Field requirement extraction utilities for Pydantic classes

Usage:

from cursus.api.factory import DAGConfigFactory

# Create factory from DAG factory = DAGConfigFactory(dag)

# Get base configuration requirements base_requirements = factory.get_base_config_requirements()

# Set base configuration factory.set_base_config(**base_values)

# Configure steps interactively for step_name in factory.get_pending_steps():

step_requirements = factory.get_step_requirements(step_name) factory.set_step_config(step_name, **step_values)

# Generate final configurations configs = factory.generate_all_configs()

class DAGConfigFactory(dag, project_root=None, anchor_file=None)

Bases: object

Interactive factory for step-by-step pipeline configuration generation.

This class provides a user-friendly interface for creating pipeline configurations by guiding users through the process of setting base configurations and step-specific configurations in a structured workflow.

Workflow: 1. Analyze DAG to get config class mapping 2. Collect base configurations first 3. Guide user through step-specific configurations 4. Generate final config instances with inheritance

auto_configure_step_if_possible(step_name)

Auto-configure a step if it only has tier 2+ (optional) fields besides inherited fields.

This method checks if a step can be configured with just the inherited base config fields, without requiring any tier 1 (essential) step-specific fields.

Parameters:

step_name (str) – Name of the step to auto-configure

Returns:

The created config instance if auto-configuration succeeded, None otherwise

Return type:

BaseModel | None

can_auto_configure_step(step_name)

Check if a step can be auto-configured (only has tier 2+ fields besides inherited).

Parameters:

step_name (str) – Name of the step to check

Returns:

True if step can be auto-configured, False if it requires user input

Return type:

bool

configure_step_if_present(step_name, **kwargs)

Configure a step, but WARN (not raise) if it is not a DAG node — a non-silent replacement for the ubiquitous if "X" in pending_steps: set_step_config("X", ...) notebook guard.

The bare guard silently skips a mistyped/renamed step name because pending_steps is derived from the DAG nodes — hiding real typos (the campaign found leftover template cells in ~7 projects this way). This wrapper resolves the name (bare→suffixed included) and, when it matches no DAG node, logs a WARNING and returns None instead of quietly doing nothing.

generate_all_configs()

Generate final list of config instances.

Automatically configures steps that only have tier 2+ fields, then validates that all essential steps are configured before generating final instances.

Returns:

List of configured instances ready for pipeline execution

Return type:

List[BaseModel]

get_all_config_instances()

Get all validated config instances.

Returns:

Dictionary mapping step names to validated config instances

Return type:

Dict[str, BaseModel]

get_base_config_requirements()

Get base configuration requirements directly from Pydantic class definition.

Extracts field requirements directly from BasePipelineConfig Pydantic class definition.

Returns:

{

‘name’: str, # Field name ‘type’: str, # Field type as string ‘description’: str, # Field description from Pydantic Field() ‘required’: bool, # True for required fields, False for optional ‘default’: Any # Default value (only for optional fields)

}

Return type:

List of field requirement dictionaries with format

get_base_processing_config_requirements()

Get base processing configuration requirements.

Returns only the non-inherited fields specific to BaseProcessingStepConfig. Inherited fields from BasePipelineConfig can be obtained by calling get_base_config_requirements().

Uses cached field requirements computed during initialization for instant response.

Returns:

List of field requirement dictionaries for processing-specific fields

Return type:

List[Dict[str, Any]]

get_config_class_map()

Get mapping of DAG node names to config classes (not instances).

Returns:

Dictionary mapping node names to configuration classes

Return type:

Dict[str, Type[BaseModel]]

get_configuration_status()

Check which configurations have been filled in.

Returns:

Dictionary mapping configuration names to completion status

Return type:

Dict[str, bool]

get_factory_summary()

Get summary information about the factory state.

Returns:

Dictionary with factory state summary

Return type:

Dict[str, Any]

get_pending_steps()

Get list of steps that still need configuration.

Steps with only tier 2+ (optional) fields besides inherited fields are considered auto-configurable and not pending.

Returns:

List of step names that haven’t been configured yet and require user input

Return type:

List[str]

get_step_config_instance(step_name)

Get the validated config instance for a step.

Parameters:

step_name (str) – Name of the step

Returns:

The validated config instance, or None if not configured

Return type:

BaseModel | None

get_step_requirements(step_name)

Get step-specific requirements excluding inherited base config fields.

Uses cached field requirements computed during initialization for instant response. This avoids the expensive field extraction process (~200+ operations) on every call.

Parameters:

step_name (str) – Name of the step to get requirements for

Returns:

List of field requirement dictionaries for step-specific fields only

Return type:

List[Dict[str, Any]]

is_dag_step(step_name, job_type=None)

Return True if step_name (bare or suffixed, optionally + job_type) is a DAG node.

Lets callers replace the silent if "X" in pending_steps: guard with an explicit, resolution-aware check that honours the same bare→suffixed logic as set_step_config.

load_partial_state(file_path)

Load previously saved factory state.

Parameters:

file_path (str) – Path to the saved state file

save_partial_state(file_path)

Save current factory state for later restoration.

Parameters:

file_path (str) – Path to save the state file

set_base_config(**kwargs)

Set base pipeline configuration from user inputs.

Parameters:

**kwargs – Base configuration field values

set_base_processing_config(**kwargs)

Set base processing configuration from user inputs.

Parameters:

**kwargs – Base processing configuration field values

set_step_config(step_name, **kwargs)

Set configuration for a specific step with immediate validation.

Creates and validates the config instance immediately using the proper from_base_config pattern, providing early feedback to users.

Parameters:

• step_name (str) – Name of the step to configure

• **kwargs – Step-specific configuration field values

Returns:

The created and validated config instance

Raises:

ValueError – If configuration is invalid or prerequisites not met

Return type:

BaseModel

update_step_config(step_name, **kwargs)

Update existing step configuration with new values.

Parameters:

• step_name (str) – Name of the step to update

• **kwargs – New configuration values to merge

Returns:

Updated and validated config instance

Raises:

ValueError – If step not configured yet or update fails

Return type:

BaseModel

validate_dag_config_alignment(raise_on_error=True)

Assert the DAG↔config invariant: every DAG node resolves to a config whose derived save-key equals the node name, and every configured instance keys back to a DAG node.

This catches the class of latent drift the multi-pipeline validation campaign found — a DAG node whose step TYPE has no matching config (e.g. DAG BedrockBatchProcessing_training vs config class BedrockProcessing), which otherwise only surfaces at pipeline-compile time as an opaque “no config for node”. We check it at generate/save time instead.

The saved config key is {registry_step_name}[_{job_type}] (see config _derive_step_name); it MUST equal the DAG node key the factory stored the instance under. A mismatch means the configured config class is the wrong step TYPE for that node.

Parameters:

raise_on_error (bool) – raise ValueError listing all mismatches (default); if False, return them.

Returns:

List of human-readable mismatch strings (empty when aligned).

Return type:

List[str]

class ConfigClassMapper

Bases: object

Maps DAG nodes to configuration classes using existing registry system.

This class provides the bridge between DAG structure and configuration classes, enabling automatic discovery of which configuration class should be used for each step in a pipeline DAG.

get_available_config_classes()

Get all available configuration classes from the registry.

Returns:

Dictionary of available configuration classes

Return type:

Dict[str, Type[BaseModel]]

map_dag_to_config_classes(dag)

Map DAG node names to configuration classes (not instances).

Parameters:

dag – Pipeline DAG object with nodes to map

Returns:

Dictionary mapping node names to configuration classes

Return type:

Dict[str, Type[BaseModel]]

register_manual_mapping(node_name, config_class)

Manually register a mapping between node name and configuration class.

This is useful for cases where automatic resolution fails or for custom configuration classes not in the registry.

Parameters:

• node_name (str) – DAG node name

• config_class (Type[BaseModel]) – Configuration class to map to the node

resolve_node_to_config_class(node_name)

Resolve a single DAG node to its configuration class.

Parameters:

node_name (str) – Name of the DAG node to resolve

Returns:

Configuration class for the node, or None if not found

Return type:

Type[BaseModel] | None

validate_mapping(config_class_map)

Validate that all mapped configuration classes are valid Pydantic models.

Parameters:

config_class_map (Dict[str, Type[BaseModel]]) – Dictionary of node name to config class mappings

Returns:

Dictionary of validation errors (empty if all valid)

Return type:

Dict[str, str]

class ConfigurationGenerator(base_config, base_processing_config=None)

Bases: object

Generates final configuration instances with base config inheritance.

This class handles the complex task of combining base pipeline configurations with step-specific configurations, ensuring proper inheritance and validation.

generate_all_instances(config_class_map, step_configs)

Generate all configuration instances with proper inheritance.

Parameters:

• config_class_map (Dict[str, Type[BaseModel]]) – Mapping of step names to configuration classes

• step_configs (Dict[str, Dict[str, Any]]) – Step-specific configuration values

Returns:

List of configured instances ready for pipeline execution

Return type:

List[BaseModel]

generate_config_instance(config_class, step_inputs)

Generate config instance using base config inheritance.

Parameters:

• config_class (Type[BaseModel]) – Configuration class to instantiate

• step_inputs (Dict[str, Any]) – Step-specific input values

Returns:

Configured instance with base config inheritance applied

Return type:

BaseModel

get_config_summary(configs)

Get summary information about generated configurations.

Parameters:

configs (List[BaseModel]) – List of generated configuration instances

Returns:

Dictionary with summary information for each config

Return type:

Dict[str, Dict[str, Any]]

validate_generated_configs(configs)

Validate generated configuration instances.

Parameters:

configs (List[BaseModel]) – List of generated configuration instances

Returns:

Dictionary mapping config class names to validation error lists

Return type:

Dict[str, List[str]]

exception ConfigurationIncompleteError

Bases: Exception

Exception raised when essential configuration fields are missing.

extract_field_requirements(config_class)

Extract field requirements directly from Pydantic V2 class definition.

Parameters:

config_class (Type[BaseModel]) – Pydantic V2 model class to extract fields from

Returns:

{

‘name’: str, # Field name ‘type’: str, # Field type as string ‘description’: str, # Field description from Pydantic Field() ‘required’: bool, # True for required fields, False for optional ‘default’: Any # Default value (only for optional fields)

}

Return type:

List of field requirement dictionaries with format

print_field_requirements(requirements)

Print field requirements in user-friendly format.

Parameters:

requirements (List[Dict[str, Any]]) – List of field requirement dictionaries

categorize_field_requirements(requirements)

Categorize field requirements into required and optional groups.

Parameters:

requirements (List[Dict[str, Any]]) – List of field requirement dictionaries

Returns:

Dictionary with ‘required’ and ‘optional’ keys containing respective field lists

Return type:

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

Modules

config_class_mapper	Configuration Class Mapper
configuration_generator	Configuration Generator
dag_config_factory	DAG Configuration Factory
field_extractor	Field Requirement Extraction Utilities