cursus.steps.configs.utils

Configuration utility functions for merging, saving, and loading multiple Pydantic configs.

IMPORTANT: This module is maintained for backward compatibility. For new code, please import directly from src.config_field_manager:

from …config_field_manager import merge_and_save_configs, load_configs

This module provides a high-level API for configuration management, leveraging the optimized implementation in src.config_field_manager while maintaining backward compatibility with existing code.

detect_config_classes_from_json(config_path)[source]

Fallback implementation that simply calls build_complete_config_classes.

class CategoryType(*values)[source]

Bases: Enum

SHARED = 1
SPECIFIC = 2
serialize_config(config)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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