cursus.api.factory.dag_config_factory

DAG Configuration Factory

This module provides the main DAGConfigFactory class for interactive pipeline configuration generation. It orchestrates the step-by-step workflow for collecting user inputs and generating complete pipeline configurations.

Key Components: - DAGConfigFactory: Main interactive factory class - ConfigurationIncompleteError: Exception for incomplete configurations - Interactive workflow management and validation

exception ConfigurationIncompleteError

Bases: Exception

Exception raised when essential configuration fields are missing.

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

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

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

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]

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

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

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

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.

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.

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

get_configuration_status()

Check which configurations have been filled in.

Returns:

Dictionary mapping configuration names to completion status

Return type:

Dict[str, bool]

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]

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_factory_summary()

Get summary information about the factory state.

Returns:

Dictionary with factory state summary

Return type:

Dict[str, Any]

save_partial_state(file_path)

Save current factory state for later restoration.

Parameters:

file_path (str) – Path to save the state file

load_partial_state(file_path)

Load previously saved factory state.

Parameters:

file_path (str) – Path to the saved state file

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

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_all_config_instances()

Get all validated config instances.

Returns:

Dictionary mapping step names to validated config instances

Return type:

Dict[str, BaseModel]