cursus.api.factory.config_class_mapper

Configuration Class Mapper

This module provides mapping functionality between DAG nodes and their corresponding configuration classes using the existing registry system. It handles the resolution of DAG node names to Pydantic configuration classes.

Key Components: - ConfigClassMapper: Maps DAG nodes to configuration classes using registry system - Registry integration for automatic DAG node-to-config mapping - Fallback mechanisms for cases where registry is unavailable

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.

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

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

get_available_config_classes()

Get all available configuration classes from the registry.

Returns:

Dictionary of available 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

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 ManualConfigClassMapper(manual_mappings=None)

Bases: ConfigClassMapper

Manual configuration class mapper for cases where registry is not available.

This mapper allows explicit registration of node-to-config-class mappings without relying on the registry system.

add_mapping(node_name, config_class)

Add a manual mapping between node name and configuration class.

Parameters:

• node_name (str) – DAG node name

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

add_mappings(mappings)

Add multiple manual mappings at once.

Parameters:

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