"""
Unified Config Manager - Single integrated component replacing redundant data structures.
This module provides a unified interface that replaces three separate systems:
- ConfigClassStore (already migrated to step catalog adapter)
- TierRegistry (eliminated - uses config class methods)
- CircularReferenceTracker (simplified to minimal tier-aware tracking)
Total Reduction: 950 lines → 120 lines (87% reduction)
"""
import logging
from typing import Any, Dict, List, Optional, Set, Type
from pathlib import Path
from pydantic import BaseModel
logger = logging.getLogger(__name__)
[docs]
class SimpleTierAwareTracker:
"""
Simple tier-aware circular reference tracking.
Replaces the complex CircularReferenceTracker (600+ lines) with minimal
tracking based on three-tier architecture constraints.
"""
def __init__(self):
"""Initialize simple tracking with visited set."""
self.visited: Set[int] = set()
self.processing_stack: List[str] = []
self.max_depth = 50 # Reasonable limit for config objects
[docs]
def enter_object(self, obj: Any, field_name: Optional[str] = None) -> bool:
"""
Check if object creates circular reference.
Args:
obj: Object being processed
field_name: Name of field containing object
Returns:
bool: True if circular reference detected
"""
# Check depth limit
if len(self.processing_stack) >= self.max_depth:
logger.warning(f"Max depth {self.max_depth} exceeded at field {field_name}")
return True
# Simple ID-based tracking for dictionaries with type info
if isinstance(obj, dict) and "__model_type__" in obj:
obj_id = id(obj)
if obj_id in self.visited:
logger.warning(f"Circular reference detected in {field_name}")
return True
self.visited.add(obj_id)
# Track processing stack for depth management
self.processing_stack.append(field_name or "unknown")
return False
[docs]
def exit_object(self) -> None:
"""Exit current object processing."""
if self.processing_stack:
self.processing_stack.pop()
[docs]
def reset(self) -> None:
"""Reset tracker state."""
self.visited.clear()
self.processing_stack.clear()
[docs]
class UnifiedConfigManager:
"""
Single integrated component replacing three separate systems.
Replaces:
- ConfigClassStore: Uses step catalog integration
- TierRegistry: Uses config classes' own categorize_fields() methods
- CircularReferenceTracker: Simple tier-aware tracking
Total Reduction: 950 lines → 120 lines (87% reduction)
"""
def __init__(self, workspace_dirs: Optional[List[str]] = None):
"""
Initialize unified config manager.
Args:
workspace_dirs: List of workspace directories for step catalog integration
"""
self.workspace_dirs = workspace_dirs or []
self.simple_tracker = SimpleTierAwareTracker()
self._step_catalog = None
@property
def step_catalog(self):
"""Lazy-load step catalog to avoid import issues."""
if self._step_catalog is None:
try:
from ...step_catalog import StepCatalog
# Use workspace_dirs directly as step catalog expects
self._step_catalog = StepCatalog(workspace_dirs=self.workspace_dirs)
except ImportError:
logger.warning("Step catalog not available, using fallback")
self._step_catalog = None
return self._step_catalog
[docs]
def get_config_classes(
self, project_id: Optional[str] = None
) -> Dict[str, Type[BaseModel]]:
"""
Get config classes using step catalog integration.
Replaces ConfigClassStore functionality.
Args:
project_id: Optional project ID for workspace-specific discovery
Returns:
Dictionary mapping class names to class types
"""
try:
if self.step_catalog:
discovered_classes = self.step_catalog.build_complete_config_classes(
project_id
)
logger.debug(
f"Discovered {len(discovered_classes)} config classes via step catalog"
)
return discovered_classes
else:
# Fallback to direct import
from ...step_catalog.config_discovery import ConfigAutoDiscovery
from ...step_catalog import StepCatalog
# ✅ CORRECT: Use StepCatalog's package root detection
# Reuse existing _find_package_root logic from StepCatalog
temp_catalog = StepCatalog(workspace_dirs=None)
package_root = temp_catalog.package_root
config_discovery = ConfigAutoDiscovery(
package_root=package_root, # Cursus package location (from StepCatalog)
workspace_dirs=self.workspace_dirs, # User workspace directories
)
discovered_classes = config_discovery.build_complete_config_classes(
project_id
)
logger.debug(
f"Discovered {len(discovered_classes)} config classes via ConfigAutoDiscovery"
)
return discovered_classes
except Exception as e:
logger.error(f"Config class discovery failed: {e}")
# Final fallback - return basic classes
return self._get_basic_config_classes()
[docs]
def get_field_tiers(self, config_instance: BaseModel) -> Dict[str, List[str]]:
"""
Get field tier information using config's own methods.
Replaces TierRegistry functionality by using config classes'
own categorize_fields() methods.
Args:
config_instance: Config instance to categorize
Returns:
Dictionary mapping tier names to field lists
"""
try:
# Use config's own categorize_fields method if available
if hasattr(config_instance, "categorize_fields"):
return config_instance.categorize_fields()
else:
# Fallback to basic categorization
logger.warning(
f"Config {type(config_instance).__name__} has no categorize_fields method"
)
return self._basic_field_categorization(config_instance)
except Exception as e:
logger.error(f"Field categorization failed: {e}")
return self._basic_field_categorization(config_instance)
[docs]
def serialize_with_tier_awareness(self, obj: Any) -> Any:
"""
Serialize object with simple tier-aware circular reference tracking.
Replaces complex CircularReferenceTracker with minimal tracking.
Args:
obj: Object to serialize
Returns:
Serialized object
"""
self.simple_tracker.reset()
return self._serialize_recursive(obj)
def _serialize_recursive(self, obj: Any, field_name: Optional[str] = None) -> Any:
"""
Recursively serialize object with circular reference protection.
Args:
obj: Object to serialize
field_name: Name of current field
Returns:
Serialized object
"""
# Check for circular reference
if self.simple_tracker.enter_object(obj, field_name):
return f"<circular_reference_to_{field_name}>"
try:
# Handle different object types
if isinstance(obj, BaseModel):
# Pydantic model - use model_dump
result = obj.model_dump()
elif isinstance(obj, dict):
# Dictionary - serialize recursively
result = {
k: self._serialize_recursive(
v, f"{field_name}.{k}" if field_name else k
)
for k, v in obj.items()
}
elif isinstance(obj, (list, tuple)):
# List/tuple - serialize elements
result = [
self._serialize_recursive(
item, f"{field_name}[{i}]" if field_name else f"[{i}]"
)
for i, item in enumerate(obj)
]
else:
# Primitive type - return as-is
result = obj
return result
finally:
self.simple_tracker.exit_object()
def _get_basic_config_classes(self) -> Dict[str, Type[BaseModel]]:
"""
Get basic config classes as final fallback.
Returns:
Dictionary with basic config classes
"""
try:
from ...core.base.config_base import BasePipelineConfig
from ...steps.configs.config_processing_step_base import (
ProcessingStepConfigBase,
)
from ...core.base.hyperparameters_base import ModelHyperparameters
return {
"BasePipelineConfig": BasePipelineConfig,
"ProcessingStepConfigBase": ProcessingStepConfigBase,
"ModelHyperparameters": ModelHyperparameters,
}
except ImportError as e:
logger.error(f"Could not import basic config classes: {e}")
return {}
def _basic_field_categorization(
self, config_instance: BaseModel
) -> Dict[str, List[str]]:
"""
Basic field categorization fallback.
Args:
config_instance: Config instance to categorize
Returns:
Basic field categorization
"""
fields = list(config_instance.model_fields.keys())
# Simple categorization based on field names
essential_fields = []
system_fields = []
derived_fields = []
for field in fields:
if any(
keyword in field.lower()
for keyword in ["name", "id", "region", "field_list"]
):
essential_fields.append(field)
elif any(
keyword in field.lower()
for keyword in ["instance", "framework", "entry_point"]
):
system_fields.append(field)
else:
derived_fields.append(field)
return {
"essential": essential_fields,
"system": system_fields,
"derived": derived_fields,
}
def _get_basic_form_fields(
self, config_class_name: str, config_class: Optional[Type[BaseModel]] = None
) -> List[Dict[str, Any]]:
"""
Basic form field extraction fallback.
Args:
config_class_name: Name of the configuration class
config_class: Optional config class
Returns:
Basic field definitions
"""
if config_class is None:
config_classes = self.get_config_classes()
config_class = config_classes.get(config_class_name)
if not config_class:
logger.warning(f"Config class {config_class_name} not found")
return []
fields = []
# Extract basic field information from Pydantic model
for field_name, field_info in config_class.model_fields.items():
try:
field_type = field_info.annotation
field_required = field_info.is_required()
field_default = getattr(field_info, "default", None)
field_description = getattr(field_info, "description", "")
fields.append(
{
"name": field_name,
"type": str(field_type),
"required": field_required,
"default": field_default,
"description": field_description,
"tier": "essential" if field_required else "system",
}
)
except Exception as e:
logger.debug(f"Could not extract field {field_name}: {e}")
continue
return fields
def _verify_essential_structure(self, merged: Dict[str, Any]) -> None:
"""
Simplified verification method covering critical requirements only.
Phase 1 Day 3-4 optimization: Single verification method replacing
multiple overlapping verification methods (60% code reduction).
Args:
merged: Merged configuration structure to verify
Raises:
ValueError: If essential structure requirements are not met
"""
# Verify essential structure (shared/specific sections)
if not isinstance(merged, dict):
raise ValueError("Merged configuration must be a dictionary")
if "shared" not in merged:
raise ValueError(
"Missing required 'shared' section in merged configuration"
)
if "specific" not in merged:
raise ValueError(
"Missing required 'specific' section in merged configuration"
)
# Verify shared section is a dictionary
if not isinstance(merged["shared"], dict):
raise ValueError("'shared' section must be a dictionary")
# Verify specific section is a dictionary
if not isinstance(merged["specific"], dict):
raise ValueError("'specific' section must be a dictionary")
# Verify critical field placement (mutual exclusivity)
shared_fields = set(merged["shared"].keys())
for step_name, step_fields in merged["specific"].items():
if not isinstance(step_fields, dict):
raise ValueError(f"Step '{step_name}' fields must be a dictionary")
step_field_names = set(step_fields.keys())
# Check for field conflicts between shared and specific
conflicts = shared_fields.intersection(step_field_names)
if conflicts:
logger.warning(
f"Field conflicts detected between shared and specific in step '{step_name}': {conflicts}"
)
# Note: This is a warning, not an error, as some overlap might be intentional
logger.debug(
f"Structure verification passed: {len(merged['shared'])} shared fields, {len(merged['specific'])} specific steps"
)
[docs]
def save(
self,
config_list: List[Any],
output_file: str,
processing_step_config_base_class: Optional[type] = None,
) -> Dict[str, Any]:
"""
Save merged configuration to a file using UnifiedConfigManager.
Includes optimized verification from Phase 1 Day 3-4 improvements.
Args:
config_list: List of configuration objects to merge and save
output_file: Path to output file
processing_step_config_base_class: Optional base class for processing steps
Returns:
dict: Merged configuration structure
"""
import json
import os
from datetime import datetime
from .step_catalog_aware_categorizer import (
StepCatalogAwareConfigFieldCategorizer,
)
from .type_aware_config_serializer import TypeAwareConfigSerializer
# Ensure directory exists
os.makedirs(os.path.dirname(os.path.abspath(output_file)), exist_ok=True)
# Use step catalog aware categorizer
categorizer = StepCatalogAwareConfigFieldCategorizer(
config_list, processing_step_config_base_class
)
# Get categorized fields
categorized = categorizer.get_categorized_fields()
merged = {"shared": categorized["shared"], "specific": categorized["specific"]}
# Apply optimized verification (Phase 1 Day 3-4 improvement)
self._verify_essential_structure(merged)
# Create metadata
config_types = {}
serializer = TypeAwareConfigSerializer()
for cfg in config_list:
step_name = serializer.generate_step_name(cfg)
class_name = cfg.__class__.__name__
config_types[step_name] = class_name
field_sources = categorizer.get_field_sources()
metadata = {
"created_at": datetime.now().isoformat(),
"config_types": config_types,
"field_sources": field_sources,
}
# Create output structure
output = {"metadata": metadata, "configuration": merged}
# Save to file
logger.info(f"Saving merged configuration to {output_file}")
with open(output_file, "w") as f:
json.dump(output, f, indent=2, sort_keys=True)
logger.info(f"Successfully saved merged configuration to {output_file}")
return merged
[docs]
def load(
self, input_file: str, config_classes: Optional[Dict[str, type]] = None
) -> Dict[str, Any]:
"""
Load a merged configuration from a file using UnifiedConfigManager.
Args:
input_file: Path to input file
config_classes: Optional mapping of class names to class objects
Returns:
dict: Loaded configuration structure
"""
import json
import os
from .type_aware_config_serializer import TypeAwareConfigSerializer
logger.info(f"Loading configuration from {input_file}")
if not os.path.exists(input_file):
raise FileNotFoundError(f"Configuration file not found: {input_file}")
# Load JSON file
with open(input_file, "r") as f:
file_data = json.load(f)
# Handle both old and new formats
if "configuration" in file_data and isinstance(
file_data["configuration"], dict
):
data = file_data["configuration"]
else:
data = file_data
# Use config classes from UnifiedConfigManager if not provided
if config_classes is None:
config_classes = self.get_config_classes()
# Create serializer
serializer = TypeAwareConfigSerializer(config_classes=config_classes)
# Process into simplified structure
result: Dict[str, Any] = {"shared": {}, "specific": {}}
# Deserialize shared fields
if "shared" in data:
for field, value in data["shared"].items():
result["shared"][field] = serializer.deserialize(value)
# Deserialize specific fields
if "specific" in data:
for step, fields in data["specific"].items():
if step not in result["specific"]:
result["specific"][step] = {}
for field, value in fields.items():
result["specific"][step][field] = serializer.deserialize(value)
logger.info(f"Successfully loaded configuration from {input_file}")
return result
# Cache of manager instances keyed by their workspace_dirs. Previously a single global was
# cached on first call, so a later call with DIFFERENT workspace_dirs silently returned the
# stale (wrong-context) manager. Keying by the dirs makes each distinct context get its own.
_unified_managers: "Dict[tuple, UnifiedConfigManager]" = {}
def _workspace_dirs_key(workspace_dirs: Optional[List[str]]) -> tuple:
"""Normalize workspace_dirs into a stable, hashable cache key."""
return tuple(str(d) for d in (workspace_dirs or []))
[docs]
def get_unified_config_manager(
workspace_dirs: Optional[List[str]] = None,
) -> UnifiedConfigManager:
"""
Get a unified config manager instance, cached per ``workspace_dirs``.
Args:
workspace_dirs: List of workspace directories for step catalog integration
Returns:
UnifiedConfigManager instance (one per distinct workspace_dirs key)
"""
key = _workspace_dirs_key(workspace_dirs)
manager = _unified_managers.get(key)
if manager is None:
manager = UnifiedConfigManager(workspace_dirs)
_unified_managers[key] = manager
return manager
[docs]
def reset_unified_config_manager_cache() -> None:
"""Clear the per-workspace_dirs manager cache (mainly for tests / hot-reload)."""
_unified_managers.clear()