Coverage for openhcs/config_framework/lazy_factory.py: 70.1%
445 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-04 02:09 +0000
1"""Generic lazy dataclass factory using flexible resolution."""
3# Standard library imports
4import dataclasses
5import logging
6import re
7import sys
8from abc import ABCMeta
9from dataclasses import dataclass, fields, is_dataclass, make_dataclass, MISSING, field
10from typing import Any, Callable, Dict, List, Optional, Tuple, Type, TypeVar, Union
12# OpenHCS imports
13from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService
14from openhcs.core.auto_register_meta import AutoRegisterMeta, RegistryConfig
15# Note: dual_axis_resolver_recursive and lazy_placeholder imports kept inline to avoid circular imports
18# Type registry for lazy dataclass to base class mapping
19_lazy_type_registry: Dict[Type, Type] = {}
21# Cache for lazy classes to prevent duplicate creation
22_lazy_class_cache: Dict[str, Type] = {}
25# ContextEventCoordinator removed - replaced with contextvars-based context system
30def register_lazy_type_mapping(lazy_type: Type, base_type: Type) -> None:
31 """Register mapping between lazy dataclass type and its base type."""
32 _lazy_type_registry[lazy_type] = base_type
35def get_base_type_for_lazy(lazy_type: Type) -> Optional[Type]:
36 """Get the base type for a lazy dataclass type."""
37 return _lazy_type_registry.get(lazy_type)
39# Optional imports (handled gracefully)
40try:
41 from PyQt6.QtWidgets import QApplication
42 HAS_PYQT = True
43except ImportError:
44 QApplication = None
45 HAS_PYQT = False
47logger = logging.getLogger(__name__)
50# Constants for lazy configuration system - simplified from class to module-level
51MATERIALIZATION_DEFAULTS_PATH = "materialization_defaults"
52RESOLVE_FIELD_VALUE_METHOD = "_resolve_field_value"
53GET_ATTRIBUTE_METHOD = "__getattribute__"
54TO_BASE_CONFIG_METHOD = "to_base_config"
55WITH_DEFAULTS_METHOD = "with_defaults"
56WITH_OVERRIDES_METHOD = "with_overrides"
57LAZY_FIELD_DEBUG_TEMPLATE = "LAZY FIELD CREATION: {field_name} - original={original_type}, has_default={has_default}, final={final_type}"
59LAZY_CLASS_NAME_PREFIX = "Lazy"
61# Legacy helper functions removed - new context system handles all resolution
64# Functional fallback strategies
65def _get_raw_field_value(obj: Any, field_name: str) -> Any:
66 """
67 Get raw field value bypassing lazy property getters to prevent infinite recursion.
69 Uses object.__getattribute__() to access stored values directly without triggering
70 lazy resolution, which would create circular dependencies in the resolution chain.
72 Args:
73 obj: Object to get field from
74 field_name: Name of field to access
76 Returns:
77 Raw field value or None if field doesn't exist
79 Raises:
80 AttributeError: If field doesn't exist (fail-loud behavior)
81 """
82 try:
83 return object.__getattribute__(obj, field_name)
84 except AttributeError:
85 return None
88@dataclass(frozen=True)
89class LazyMethodBindings:
90 """Declarative method bindings for lazy dataclasses."""
92 @staticmethod
93 def create_resolver() -> Callable[[Any, str], Any]:
94 """Create field resolver method using new pure function interface."""
95 from openhcs.config_framework.dual_axis_resolver import resolve_field_inheritance
96 from openhcs.config_framework.context_manager import current_temp_global, extract_all_configs
98 def _resolve_field_value(self, field_name: str) -> Any:
99 # Get current context from contextvars
100 try:
101 current_context = current_temp_global.get()
102 # Extract available configs from current context
103 available_configs = extract_all_configs(current_context)
105 # Use pure function for resolution
106 return resolve_field_inheritance(self, field_name, available_configs)
107 except LookupError:
108 # No context available - return None (fail-loud approach)
109 logger.debug(f"No context available for resolving {type(self).__name__}.{field_name}")
110 return None
112 return _resolve_field_value
114 @staticmethod
115 def create_getattribute() -> Callable[[Any, str], Any]:
116 """Create lazy __getattribute__ method using new context system."""
117 from openhcs.config_framework.dual_axis_resolver import resolve_field_inheritance, _has_concrete_field_override
118 from openhcs.config_framework.context_manager import current_temp_global, extract_all_configs
120 def _find_mro_concrete_value(base_class, name):
121 """Extract common MRO traversal pattern."""
122 return next((getattr(cls, name) for cls in base_class.__mro__
123 if _has_concrete_field_override(cls, name)), None)
125 def _try_global_context_value(self, base_class, name):
126 """Extract global context resolution logic using new pure function interface."""
127 if not hasattr(self, '_global_config_type'):
128 return None
130 # Get current context from contextvars
131 try:
132 current_context = current_temp_global.get()
133 # Extract available configs from current context
134 available_configs = extract_all_configs(current_context)
136 # Use pure function for resolution
137 resolved_value = resolve_field_inheritance(self, name, available_configs)
138 if resolved_value is not None:
139 return resolved_value
140 except LookupError:
141 # No context available - fall back to MRO
142 pass
144 # Fallback to MRO concrete value
145 return _find_mro_concrete_value(base_class, name)
147 def __getattribute__(self: Any, name: str) -> Any:
148 """
149 Three-stage resolution using new context system.
151 Stage 1: Check instance value
152 Stage 2: Simple field path lookup in current scope's merged config
153 Stage 3: Inheritance resolution using same merged context
154 """
155 # Stage 1: Get instance value
156 value = object.__getattribute__(self, name)
157 if value is not None or name not in {f.name for f in fields(self.__class__)}:
158 return value
160 # Stage 2: Simple field path lookup in current scope's merged global
161 try:
162 current_context = current_temp_global.get()
163 if current_context is not None: 163 ↛ 181line 163 didn't jump to line 181 because the condition on line 163 was always true
164 # Get the config type name for this lazy class
165 config_field_name = getattr(self, '_config_field_name', None)
166 if config_field_name: 166 ↛ 181line 166 didn't jump to line 181 because the condition on line 166 was always true
167 try:
168 config_instance = getattr(current_context, config_field_name)
169 if config_instance is not None: 169 ↛ 181line 169 didn't jump to line 181 because the condition on line 169 was always true
170 resolved_value = getattr(config_instance, name)
171 if resolved_value is not None:
172 return resolved_value
173 except AttributeError:
174 # Field doesn't exist in merged config, continue to inheritance
175 pass
176 except LookupError:
177 # No context available, continue to inheritance
178 pass
180 # Stage 3: Inheritance resolution using same merged context
181 try:
182 current_context = current_temp_global.get()
183 available_configs = extract_all_configs(current_context)
184 resolved_value = resolve_field_inheritance(self, name, available_configs)
186 if resolved_value is not None:
187 return resolved_value
189 # For nested dataclass fields, return lazy instance
190 field_obj = next((f for f in fields(self.__class__) if f.name == name), None)
191 if field_obj and is_dataclass(field_obj.type): 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 return field_obj.type()
194 return None
196 except LookupError:
197 # No context available - fallback to MRO concrete values
198 return _find_mro_concrete_value(get_base_type_for_lazy(self.__class__), name)
199 return __getattribute__
201 @staticmethod
202 def create_to_base_config(base_class: Type) -> Callable[[Any], Any]:
203 """Create base config converter method."""
204 def to_base_config(self):
205 # CRITICAL FIX: Use object.__getattribute__ to preserve raw None values
206 # getattr() triggers lazy resolution, converting None to static defaults
207 # None values must be preserved for dual-axis inheritance to work correctly
208 #
209 # Context: to_base_config() is called DURING config_context() setup (line 124 in context_manager.py)
210 # If we use getattr() here, it triggers resolution BEFORE the context is fully set up,
211 # causing resolution to use the wrong/stale context and losing the GlobalPipelineConfig base.
212 # We must extract raw None values here, let config_context() merge them into the hierarchy,
213 # and THEN resolution happens later with the properly built context.
214 field_values = {f.name: object.__getattribute__(self, f.name) for f in fields(self)}
215 return base_class(**field_values)
216 return to_base_config
218 @staticmethod
219 def create_class_methods() -> Dict[str, Any]:
220 """Create class-level utility methods."""
221 return {
222 WITH_DEFAULTS_METHOD: classmethod(lambda cls: cls()),
223 WITH_OVERRIDES_METHOD: classmethod(lambda cls, **kwargs: cls(**kwargs))
224 }
227class LazyDataclassFactory:
228 """Generic factory for creating lazy dataclasses with flexible resolution."""
234 @staticmethod
235 def _introspect_dataclass_fields(base_class: Type, debug_template: str, global_config_type: Type = None, parent_field_path: str = None, parent_instance_provider: Optional[Callable[[], Any]] = None) -> List[Tuple[str, Type, None]]:
236 """
237 Introspect dataclass fields for lazy loading.
239 Converts nested dataclass fields to lazy equivalents and makes fields Optional
240 if they lack defaults. Complex logic handles type unwrapping and lazy nesting.
241 """
242 base_fields = fields(base_class)
243 lazy_field_definitions = []
245 for field in base_fields:
246 # Check if field already has Optional type
247 origin = getattr(field.type, '__origin__', None)
248 is_already_optional = (origin is Union and
249 type(None) in getattr(field.type, '__args__', ()))
251 # Check if field has default value or factory
252 has_default = (field.default is not MISSING or
253 field.default_factory is not MISSING)
255 # Check if field type is a dataclass that should be made lazy
256 field_type = field.type
257 if is_dataclass(field.type):
258 # SIMPLIFIED: Create lazy version using simple factory
259 lazy_nested_type = LazyDataclassFactory.make_lazy_simple(
260 base_class=field.type,
261 lazy_class_name=f"Lazy{field.type.__name__}"
262 )
263 field_type = lazy_nested_type
264 logger.debug(f"Created lazy class for {field.name}: {field.type} -> {lazy_nested_type}")
266 # Complex type logic: make Optional if no default, preserve existing Optional types
267 if is_already_optional or not has_default:
268 final_field_type = Union[field_type, type(None)] if not is_already_optional else field_type
269 else:
270 final_field_type = field_type
272 # CRITICAL FIX: Create default factory for Optional dataclass fields
273 # This eliminates the need for field introspection and ensures UI always has instances to render
274 # CRITICAL: Always preserve metadata from original field (e.g., ui_hidden flag)
275 if (is_already_optional or not has_default) and is_dataclass(field.type): 275 ↛ 279line 275 didn't jump to line 279 because the condition on line 275 was never true
276 # For Optional dataclass fields, create default factory that creates lazy instances
277 # This ensures the UI always has nested lazy instances to render recursively
278 # CRITICAL: field_type is already the lazy type, so use it directly
279 field_def = (field.name, final_field_type, dataclasses.field(default_factory=field_type, metadata=field.metadata))
280 elif field.metadata:
281 # For fields with metadata but no dataclass default factory, create a Field object to preserve metadata
282 # We need to replicate the original field's default behavior
283 if field.default is not MISSING: 283 ↛ 284line 283 didn't jump to line 284 because the condition on line 283 was never true
284 field_def = (field.name, final_field_type, dataclasses.field(default=field.default, metadata=field.metadata))
285 elif field.default_factory is not MISSING: 285 ↛ 289line 285 didn't jump to line 289 because the condition on line 285 was always true
286 field_def = (field.name, final_field_type, dataclasses.field(default_factory=field.default_factory, metadata=field.metadata))
287 else:
288 # Field has metadata but no default - use MISSING to indicate required field
289 field_def = (field.name, final_field_type, dataclasses.field(default=MISSING, metadata=field.metadata))
290 else:
291 # No metadata, no special handling needed
292 field_def = (field.name, final_field_type, None)
294 lazy_field_definitions.append(field_def)
296 # Debug logging with provided template (reduced to DEBUG level to reduce log pollution)
297 logger.debug(debug_template.format(
298 field_name=field.name,
299 original_type=field.type,
300 has_default=has_default,
301 final_type=final_field_type
302 ))
304 return lazy_field_definitions
306 @staticmethod
307 def _create_lazy_dataclass_unified(
308 base_class: Type,
309 instance_provider: Callable[[], Any],
310 lazy_class_name: str,
311 debug_template: str,
312 use_recursive_resolution: bool = False,
313 fallback_chain: Optional[List[Callable[[str], Any]]] = None,
314 global_config_type: Type = None,
315 parent_field_path: str = None,
316 parent_instance_provider: Optional[Callable[[], Any]] = None
317 ) -> Type:
318 """
319 Create lazy dataclass with declarative configuration.
321 Core factory method that creates lazy dataclass with introspected fields,
322 binds resolution methods, and registers type mappings. Complex orchestration
323 of field analysis, method binding, and class creation.
324 """
325 if not is_dataclass(base_class): 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true
326 raise ValueError(f"{base_class} must be a dataclass")
328 # Check cache first to prevent duplicate creation
329 cache_key = f"{base_class.__name__}_{lazy_class_name}_{id(instance_provider)}"
330 if cache_key in _lazy_class_cache: 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 return _lazy_class_cache[cache_key]
333 # ResolutionConfig system removed - dual-axis resolver handles all resolution
335 # Create lazy dataclass with introspected fields
336 # CRITICAL FIX: Avoid inheriting from classes with custom metaclasses to prevent descriptor conflicts
337 # Exception: InheritAsNoneMeta is safe to inherit from as it only modifies field defaults
338 # Exception: Classes with _inherit_as_none marker are safe even with ABCMeta (processed by @global_pipeline_config)
339 base_metaclass = type(base_class)
340 has_inherit_as_none_marker = hasattr(base_class, '_inherit_as_none') and base_class._inherit_as_none
341 has_unsafe_metaclass = (
342 (hasattr(base_class, '__metaclass__') or base_metaclass != type) and
343 base_metaclass != InheritAsNoneMeta and
344 not has_inherit_as_none_marker
345 )
347 if has_unsafe_metaclass: 347 ↛ 349line 347 didn't jump to line 349 because the condition on line 347 was never true
348 # Base class has unsafe custom metaclass - don't inherit, just copy interface
349 print(f"🔧 LAZY FACTORY: {base_class.__name__} has custom metaclass {base_metaclass.__name__}, avoiding inheritance")
350 lazy_class = make_dataclass(
351 lazy_class_name,
352 LazyDataclassFactory._introspect_dataclass_fields(
353 base_class, debug_template, global_config_type, parent_field_path, parent_instance_provider
354 ),
355 bases=(), # No inheritance to avoid metaclass conflicts
356 frozen=True
357 )
358 else:
359 # Safe to inherit from regular dataclass
360 lazy_class = make_dataclass(
361 lazy_class_name,
362 LazyDataclassFactory._introspect_dataclass_fields(
363 base_class, debug_template, global_config_type, parent_field_path, parent_instance_provider
364 ),
365 bases=(base_class,),
366 frozen=True
367 )
369 # Add constructor parameter tracking to detect user-set fields
370 original_init = lazy_class.__init__
371 def __init_with_tracking__(self, **kwargs):
372 # Track which fields were explicitly passed to constructor
373 object.__setattr__(self, '_explicitly_set_fields', set(kwargs.keys()))
374 # Store the global config type for inheritance resolution
375 object.__setattr__(self, '_global_config_type', global_config_type)
376 # Store the config field name for simple field path lookup
377 import re
378 def _camel_to_snake_local(name: str) -> str:
379 s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
380 return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
381 config_field_name = _camel_to_snake_local(base_class.__name__)
382 object.__setattr__(self, '_config_field_name', config_field_name)
383 original_init(self, **kwargs)
385 lazy_class.__init__ = __init_with_tracking__
387 # Bind methods declaratively - inline single-use method
388 method_bindings = {
389 RESOLVE_FIELD_VALUE_METHOD: LazyMethodBindings.create_resolver(),
390 GET_ATTRIBUTE_METHOD: LazyMethodBindings.create_getattribute(),
391 TO_BASE_CONFIG_METHOD: LazyMethodBindings.create_to_base_config(base_class),
392 **LazyMethodBindings.create_class_methods()
393 }
394 for method_name, method_impl in method_bindings.items():
395 setattr(lazy_class, method_name, method_impl)
397 # CRITICAL: Preserve original module for proper imports in generated code
398 # make_dataclass() sets __module__ to the caller's module (lazy_factory.py)
399 # We need to set it to the base class's original module for correct import paths
400 lazy_class.__module__ = base_class.__module__
402 # Automatically register the lazy dataclass with the type registry
403 register_lazy_type_mapping(lazy_class, base_class)
405 # Cache the created class to prevent duplicates
406 _lazy_class_cache[cache_key] = lazy_class
408 return lazy_class
414 @staticmethod
415 def make_lazy_simple(
416 base_class: Type,
417 lazy_class_name: str = None
418 ) -> Type:
419 """
420 Create lazy dataclass using new contextvars system.
422 SIMPLIFIED: No complex hierarchy providers or field path detection needed.
423 Uses new contextvars system for all resolution.
425 Args:
426 base_class: Base dataclass to make lazy
427 lazy_class_name: Optional name for the lazy class
429 Returns:
430 Generated lazy dataclass with contextvars-based resolution
431 """
432 # Generate class name if not provided
433 lazy_class_name = lazy_class_name or f"Lazy{base_class.__name__}"
435 # Simple provider that uses new contextvars system
436 def simple_provider():
437 """Simple provider using new contextvars system."""
438 return base_class() # Lazy __getattribute__ handles resolution
440 return LazyDataclassFactory._create_lazy_dataclass_unified(
441 base_class=base_class,
442 instance_provider=simple_provider,
443 lazy_class_name=lazy_class_name,
444 debug_template=f"Simple contextvars resolution for {base_class.__name__}",
445 use_recursive_resolution=False,
446 fallback_chain=[],
447 global_config_type=None,
448 parent_field_path=None,
449 parent_instance_provider=None
450 )
452 # All legacy methods removed - use make_lazy_simple() for all use cases
455# Generic utility functions for clean thread-local storage management
456def ensure_global_config_context(global_config_type: Type, global_config_instance: Any) -> None:
457 """Ensure proper thread-local storage setup for any global config type."""
458 from openhcs.config_framework.global_config import set_global_config_for_editing
459 set_global_config_for_editing(global_config_type, global_config_instance)
462# Context provider registry and metaclass for automatic registration
463CONTEXT_PROVIDERS = {}
466# Configuration for context provider registration
467_CONTEXT_PROVIDER_REGISTRY_CONFIG = RegistryConfig(
468 registry_dict=CONTEXT_PROVIDERS,
469 key_attribute='_context_type',
470 key_extractor=None, # Requires explicit _context_type
471 skip_if_no_key=True, # Skip if no _context_type set
472 secondary_registries=None,
473 log_registration=True,
474 registry_name='context provider'
475)
478class ContextProviderMeta(AutoRegisterMeta):
479 """Metaclass for automatic registration of context provider classes."""
481 def __new__(mcs, name, bases, attrs):
482 return super().__new__(mcs, name, bases, attrs,
483 registry_config=_CONTEXT_PROVIDER_REGISTRY_CONFIG)
486class ContextProvider(metaclass=ContextProviderMeta):
487 """Base class for objects that can provide context for lazy resolution."""
488 _context_type: Optional[str] = None # Override in subclasses
491def _detect_context_type(obj: Any) -> Optional[str]:
492 """
493 Detect what type of context object this is using registered providers.
495 Returns the context type name or None if not a recognized context type.
496 """
497 # Check for functions first (simple callable check)
498 if callable(obj) and hasattr(obj, '__name__'):
499 return "function"
501 # Check if object is an instance of any registered context provider
502 for context_type, provider_class in CONTEXT_PROVIDERS.items():
503 if isinstance(obj, provider_class):
504 return context_type
506 return None
509# ContextInjector removed - replaced with contextvars-based context system
514def resolve_lazy_configurations_for_serialization(data: Any) -> Any:
515 """
516 Recursively resolve lazy dataclass instances to concrete values for serialization.
518 CRITICAL: This function must be called WITHIN a config_context() block!
519 The context provides the hierarchy for lazy resolution.
521 How it works:
522 1. For lazy dataclasses: Access fields with getattr() to trigger resolution
523 2. The lazy __getattribute__ uses the active config_context() to resolve None values
524 3. Convert resolved values to base config for pickling
526 Example (from README.md):
527 with config_context(orchestrator.pipeline_config):
528 # Lazy resolution happens here via context
529 resolved_steps = resolve_lazy_configurations_for_serialization(steps)
530 """
531 # Check if this is a lazy dataclass
532 base_type = get_base_type_for_lazy(type(data))
533 if base_type is not None:
534 # This is a lazy dataclass - resolve fields using getattr() within the active context
535 # getattr() triggers lazy __getattribute__ which uses config_context() for resolution
536 resolved_fields = {}
537 for f in fields(data):
538 # CRITICAL: Use getattr() to trigger lazy resolution via context
539 # The active config_context() provides the hierarchy for resolution
540 resolved_value = getattr(data, f.name)
541 resolved_fields[f.name] = resolved_value
543 # Create base config instance with resolved values
544 resolved_data = base_type(**resolved_fields)
545 else:
546 # Not a lazy dataclass
547 resolved_data = data
549 # CRITICAL FIX: Handle step objects (non-dataclass objects with dataclass attributes)
550 step_context_type = _detect_context_type(resolved_data)
551 if step_context_type:
552 # This is a context object - inject it for its dataclass attributes
553 import inspect
554 frame = inspect.currentframe()
555 context_var_name = f"__{step_context_type}_context__"
556 frame.f_locals[context_var_name] = resolved_data
557 logger.debug(f"Injected {context_var_name} = {type(resolved_data).__name__}")
559 try:
560 # Process step attributes recursively
561 resolved_attrs = {}
562 for attr_name in dir(resolved_data):
563 if attr_name.startswith('_'):
564 continue
565 try:
566 attr_value = getattr(resolved_data, attr_name)
567 if not callable(attr_value): # Skip methods
568 logger.debug(f"Resolving {type(resolved_data).__name__}.{attr_name} = {type(attr_value).__name__}")
569 resolved_attrs[attr_name] = resolve_lazy_configurations_for_serialization(attr_value)
570 except (AttributeError, Exception):
571 continue
573 # Handle function objects specially - they can't be recreated with __new__
574 if step_context_type == "function":
575 # For functions, just process attributes for resolution but return original function
576 # The resolved config values will be stored in func plan by compiler
577 return resolved_data
579 # Create new step object with resolved attributes
580 # CRITICAL FIX: Copy all original attributes using __dict__ to preserve everything
581 new_step = type(resolved_data).__new__(type(resolved_data))
583 # Copy all attributes from the original object's __dict__
584 if hasattr(resolved_data, '__dict__'): 584 ↛ 588line 584 didn't jump to line 588 because the condition on line 584 was always true
585 new_step.__dict__.update(resolved_data.__dict__)
587 # Update with resolved config attributes (these override the originals)
588 for attr_name, attr_value in resolved_attrs.items():
589 setattr(new_step, attr_name, attr_value)
590 return new_step
591 finally:
592 if context_var_name in frame.f_locals: 592 ↛ 594line 592 didn't jump to line 594 because the condition on line 592 was always true
593 del frame.f_locals[context_var_name]
594 del frame
596 # Recursively process nested structures based on type
597 elif is_dataclass(resolved_data) and not isinstance(resolved_data, type):
598 # Process dataclass fields recursively - inline field processing pattern
599 # CRITICAL FIX: Inject parent object as context for sibling config inheritance
600 context_type = _detect_context_type(resolved_data) or "dataclass" # Default to "dataclass" for generic dataclasses
601 import inspect
602 frame = inspect.currentframe()
603 context_var_name = f"__{context_type}_context__"
604 frame.f_locals[context_var_name] = resolved_data
605 logger.debug(f"Injected {context_var_name} = {type(resolved_data).__name__}")
607 # Add debug to see which fields are being resolved
608 logger.debug(f"Resolving fields for {type(resolved_data).__name__}: {[f.name for f in fields(resolved_data)]}")
610 try:
611 resolved_fields = {}
612 for f in fields(resolved_data):
613 field_value = getattr(resolved_data, f.name)
614 logger.debug(f"Resolving {type(resolved_data).__name__}.{f.name} = {type(field_value).__name__}")
615 resolved_fields[f.name] = resolve_lazy_configurations_for_serialization(field_value)
616 return type(resolved_data)(**resolved_fields)
617 finally:
618 if context_var_name in frame.f_locals: 618 ↛ 620line 618 didn't jump to line 620 because the condition on line 618 was always true
619 del frame.f_locals[context_var_name]
620 del frame
622 elif isinstance(resolved_data, dict):
623 # Process dictionary values recursively
624 return {
625 key: resolve_lazy_configurations_for_serialization(value)
626 for key, value in resolved_data.items()
627 }
629 elif isinstance(resolved_data, (list, tuple)):
630 # Process sequence elements recursively
631 resolved_items = [resolve_lazy_configurations_for_serialization(item) for item in resolved_data]
632 return type(resolved_data)(resolved_items)
634 else:
635 # Primitive type or unknown structure - return as-is
636 return resolved_data
639# Generic dataclass editing with configurable value preservation
640T = TypeVar('T')
643def create_dataclass_for_editing(dataclass_type: Type[T], source_config: Any, preserve_values: bool = False, context_provider: Optional[Callable[[Any], None]] = None) -> T:
644 """Create dataclass for editing with configurable value preservation."""
645 if not is_dataclass(dataclass_type):
646 raise ValueError(f"{dataclass_type} must be a dataclass")
648 # Set up context if provider is given (e.g., thread-local storage)
649 if context_provider:
650 context_provider(source_config)
652 # Mathematical simplification: Convert verbose loop to unified comprehension
653 from openhcs.config_framework.placeholder import LazyDefaultPlaceholderService
654 field_values = {
655 f.name: (getattr(source_config, f.name) if preserve_values
656 else f.type() if is_dataclass(f.type) and LazyDefaultPlaceholderService.has_lazy_resolution(f.type)
657 else None)
658 for f in fields(dataclass_type)
659 }
661 return dataclass_type(**field_values)
667def rebuild_lazy_config_with_new_global_reference(
668 existing_lazy_config: Any,
669 new_global_config: Any,
670 global_config_type: Optional[Type] = None
671) -> Any:
672 """
673 Rebuild lazy config to reference new global config while preserving field states.
675 This function preserves the exact field state of the existing lazy config:
676 - Fields that are None (using lazy resolution) remain None
677 - Fields that have been explicitly set retain their concrete values
678 - Nested dataclass fields are recursively rebuilt to reference new global config
679 - The underlying global config reference is updated for None field resolution
681 Args:
682 existing_lazy_config: Current lazy config instance
683 new_global_config: New global config to reference for lazy resolution
684 global_config_type: Type of the global config (defaults to type of new_global_config)
686 Returns:
687 New lazy config instance with preserved field states and updated global reference
688 """
689 if existing_lazy_config is None:
690 return None
692 # Determine global config type
693 if global_config_type is None:
694 global_config_type = type(new_global_config)
696 # Set new global config in thread-local storage
697 ensure_global_config_context(global_config_type, new_global_config)
699 # Extract current field values without triggering lazy resolution - inline field processing pattern
700 def process_field_value(field_obj):
701 raw_value = object.__getattribute__(existing_lazy_config, field_obj.name)
703 if raw_value is not None and hasattr(raw_value, '__dataclass_fields__'):
704 try:
705 # Check if this is a concrete dataclass that should be converted to lazy
706 is_lazy = LazyDefaultPlaceholderService.has_lazy_resolution(type(raw_value))
708 if not is_lazy:
709 lazy_type = LazyDefaultPlaceholderService._get_lazy_type_for_base(type(raw_value))
711 if lazy_type:
712 # Convert concrete dataclass to lazy version while preserving ONLY non-default field values
713 # This allows fields that match class defaults to inherit from context
714 concrete_field_values = {}
715 for f in fields(raw_value):
716 field_value = object.__getattribute__(raw_value, f.name)
718 # Get the class default for this field
719 class_default = getattr(type(raw_value), f.name, None)
721 # Only preserve values that differ from class defaults
722 # This allows default values to be inherited from context
723 if field_value != class_default:
724 concrete_field_values[f.name] = field_value
726 logger.debug(f"Converting concrete {type(raw_value).__name__} to lazy version {lazy_type.__name__} for placeholder resolution")
727 return lazy_type(**concrete_field_values)
729 # If already lazy or no lazy version available, rebuild recursively
730 nested_result = rebuild_lazy_config_with_new_global_reference(raw_value, new_global_config, global_config_type)
731 return nested_result
732 except Exception as e:
733 logger.debug(f"Failed to rebuild nested config {field_obj.name}: {e}")
734 return raw_value
735 return raw_value
737 current_field_values = {f.name: process_field_value(f) for f in fields(existing_lazy_config)}
739 return type(existing_lazy_config)(**current_field_values)
742# Declarative Global Config Field Injection System
743# Moved inline imports to top-level
745# Naming configuration
746GLOBAL_CONFIG_PREFIX = "Global"
747LAZY_CONFIG_PREFIX = "Lazy"
749# Registry to accumulate all decorations before injection
750_pending_injections = {}
754class InheritAsNoneMeta(ABCMeta):
755 """
756 Metaclass that applies inherit_as_none modifications during class creation.
758 This runs BEFORE @dataclass and modifies the class definition to add
759 field overrides with None defaults for inheritance.
760 """
762 def __new__(mcs, name, bases, namespace, **kwargs):
763 # Create the class first
764 cls = super().__new__(mcs, name, bases, namespace)
766 # Check if this class should have inherit_as_none applied
767 if hasattr(cls, '_inherit_as_none') and cls._inherit_as_none:
768 # Add multiprocessing safety marker
769 cls._multiprocessing_safe = True
770 # Get explicitly defined fields (in this class's namespace)
771 explicitly_defined_fields = set()
772 if '__annotations__' in namespace:
773 for field_name in namespace['__annotations__']:
774 if field_name in namespace:
775 explicitly_defined_fields.add(field_name)
777 # Process parent classes to find fields that need None overrides
778 processed_fields = set()
779 for base in bases:
780 if hasattr(base, '__annotations__'):
781 for field_name, field_type in base.__annotations__.items():
782 if field_name in processed_fields:
783 continue
785 # Check if parent has concrete default
786 parent_has_concrete_default = False
787 if hasattr(base, field_name):
788 parent_value = getattr(base, field_name)
789 parent_has_concrete_default = parent_value is not None
791 # Add None override if needed
792 if (field_name not in explicitly_defined_fields and parent_has_concrete_default):
793 # Set the class attribute to None
794 setattr(cls, field_name, None)
796 # Ensure annotation exists
797 if not hasattr(cls, '__annotations__'):
798 cls.__annotations__ = {}
799 cls.__annotations__[field_name] = field_type
801 processed_fields.add(field_name)
802 else:
803 processed_fields.add(field_name)
805 return cls
807 def __reduce__(cls):
808 """Make classes with this metaclass pickle-safe for multiprocessing."""
809 # Filter out problematic descriptors that cause conflicts during pickle/unpickle
810 safe_dict = {}
811 for key, value in cls.__dict__.items():
812 # Skip descriptors that cause conflicts
813 if hasattr(value, '__get__') and hasattr(value, '__set__'):
814 continue # Skip data descriptors
815 if hasattr(value, '__dict__') and hasattr(value, '__class__'):
816 # Skip complex objects that might have descriptor conflicts
817 if 'descriptor' in str(type(value)).lower():
818 continue
819 # Include safe attributes
820 safe_dict[key] = value
822 # Return reconstruction using the base type (not the metaclass)
823 return (type, (cls.__name__, cls.__bases__, safe_dict))
826def create_global_default_decorator(target_config_class: Type):
827 """
828 Create a decorator factory for a specific global config class.
830 The decorator accumulates all decorations, then injects all fields at once
831 when the module finishes loading. Also creates lazy versions of all decorated configs.
832 """
833 target_class_name = target_config_class.__name__
834 if target_class_name not in _pending_injections: 834 ↛ 840line 834 didn't jump to line 840 because the condition on line 834 was always true
835 _pending_injections[target_class_name] = {
836 'target_class': target_config_class,
837 'configs_to_inject': []
838 }
840 def global_default_decorator(cls=None, *, optional: bool = False, inherit_as_none: bool = True, ui_hidden: bool = False):
841 """
842 Decorator that can be used with or without parameters.
844 Args:
845 cls: The class being decorated (when used without parentheses)
846 optional: Whether to wrap the field type with Optional (default: False)
847 inherit_as_none: Whether to set all inherited fields to None by default (default: True)
848 ui_hidden: Whether to hide from UI (apply decorator but don't inject into global config) (default: False)
849 """
850 def decorator(actual_cls):
851 # Apply inherit_as_none by modifying class BEFORE @dataclass (multiprocessing-safe)
852 if inherit_as_none: 852 ↛ 898line 852 didn't jump to line 898 because the condition on line 852 was always true
853 # Mark the class for inherit_as_none processing
854 actual_cls._inherit_as_none = True
856 # Apply inherit_as_none logic by directly modifying the class definition
857 # This must happen BEFORE @dataclass processes the class
858 explicitly_defined_fields = set()
859 if hasattr(actual_cls, '__annotations__'): 859 ↛ 869line 859 didn't jump to line 869 because the condition on line 859 was always true
860 for field_name in actual_cls.__annotations__:
861 # Check if field has a concrete default value in THIS class definition (not inherited)
862 if field_name in actual_cls.__dict__: # Only fields defined in THIS class 862 ↛ 860line 862 didn't jump to line 860 because the condition on line 862 was always true
863 field_value = actual_cls.__dict__[field_name]
864 # Only consider it explicitly defined if it has a concrete value (not None)
865 if field_value is not None:
866 explicitly_defined_fields.add(field_name)
868 # Process parent classes to find fields that need None overrides
869 processed_fields = set()
870 fields_set_to_none = set() # Track which fields were actually set to None
871 for base in actual_cls.__bases__:
872 if hasattr(base, '__annotations__'):
873 for field_name, field_type in base.__annotations__.items():
874 if field_name in processed_fields:
875 continue
877 # Set inherited fields to None (except explicitly defined ones)
878 if field_name not in explicitly_defined_fields:
879 # CRITICAL: Force the field to be seen as locally defined by @dataclass
880 # We need to ensure @dataclass processes this as a local field, not inherited
882 # 1. Set the class attribute to None
883 setattr(actual_cls, field_name, None)
884 fields_set_to_none.add(field_name)
886 # 2. Ensure annotation exists in THIS class
887 if not hasattr(actual_cls, '__annotations__'): 887 ↛ 888line 887 didn't jump to line 888 because the condition on line 887 was never true
888 actual_cls.__annotations__ = {}
889 actual_cls.__annotations__[field_name] = field_type
891 processed_fields.add(field_name)
893 # Note: We modify class attributes here, but we also need to fix the dataclass
894 # field definitions after @dataclass runs, since @dataclass processes the MRO
895 # and may use parent class field definitions instead of our modified attributes.
897 # Generate field and class names
898 field_name = _camel_to_snake(actual_cls.__name__)
899 lazy_class_name = f"{LAZY_CONFIG_PREFIX}{actual_cls.__name__}"
901 # Mark class with ui_hidden metadata for UI layer to check
902 # This allows the config to remain in the context (for lazy resolution)
903 # while being hidden from UI rendering
904 if ui_hidden:
905 actual_cls._ui_hidden = True
907 # Check if class is abstract (has unimplemented abstract methods)
908 # Abstract classes should NEVER be injected into GlobalPipelineConfig
909 # because they can't be instantiated
910 # NOTE: We need to check if the class ITSELF is abstract, not just if it inherits from ABC
911 # Concrete subclasses of abstract classes should still be injected
912 # We check for __abstractmethods__ attribute which exists even before @dataclass runs
913 # (it's set by ABCMeta when the class is created)
914 is_abstract = hasattr(actual_cls, '__abstractmethods__') and len(actual_cls.__abstractmethods__) > 0
916 # Add to pending injections for field injection
917 # Skip injection for abstract classes (they can't be instantiated)
918 # For concrete classes: inject even if ui_hidden (needed for lazy resolution context)
919 if not is_abstract:
920 _pending_injections[target_class_name]['configs_to_inject'].append({
921 'config_class': actual_cls,
922 'field_name': field_name,
923 'lazy_class_name': lazy_class_name,
924 'optional': optional, # Store the optional flag
925 'inherit_as_none': inherit_as_none, # Store the inherit_as_none flag
926 'ui_hidden': ui_hidden # Store the ui_hidden flag for field metadata
927 })
929 # Immediately create lazy version of this config (not dependent on injection)
932 lazy_class = LazyDataclassFactory.make_lazy_simple(
933 base_class=actual_cls,
934 lazy_class_name=lazy_class_name
935 )
937 # Export lazy class to config module immediately
938 config_module = sys.modules[actual_cls.__module__]
939 setattr(config_module, lazy_class_name, lazy_class)
941 # Also mark lazy class with ui_hidden metadata
942 if ui_hidden:
943 lazy_class._ui_hidden = True
945 # CRITICAL: Post-process dataclass fields after @dataclass has run
946 # This fixes the constructor behavior for inherited fields that should be None
947 if inherit_as_none and hasattr(actual_cls, '__dataclass_fields__'): 947 ↛ 950line 947 didn't jump to line 950 because the condition on line 947 was always true
948 _fix_dataclass_field_defaults_post_processing(actual_cls, fields_set_to_none)
950 return actual_cls
952 # Handle both @decorator and @decorator() usage
953 if cls is None:
954 # Called with parentheses: @decorator(optional=True)
955 return decorator
956 else:
957 # Called without parentheses: @decorator
958 return decorator(cls)
960 return global_default_decorator
963def _fix_dataclass_field_defaults_post_processing(cls: Type, fields_set_to_none: set) -> None:
964 """
965 Fix dataclass field defaults after @dataclass has processed the class.
967 This is necessary because @dataclass processes the MRO and may use parent class
968 field definitions instead of our modified class attributes. We need to ensure
969 that fields we set to None actually use None as the default in the constructor.
970 """
971 import dataclasses
973 # Store the original __init__ method
974 original_init = cls.__init__
976 def custom_init(self, **kwargs):
977 """Custom __init__ that ensures inherited fields use None defaults."""
978 # For fields that should be None, set them to None if not explicitly provided
979 for field_name in fields_set_to_none:
980 if field_name not in kwargs:
981 kwargs[field_name] = None
983 # Call the original __init__ with modified kwargs
984 original_init(self, **kwargs)
986 # Replace the __init__ method
987 cls.__init__ = custom_init
989 # Also update the field defaults for consistency
990 for field_name in fields_set_to_none:
991 if field_name in cls.__dataclass_fields__: 991 ↛ 990line 991 didn't jump to line 990 because the condition on line 991 was always true
992 # Get the field object
993 field_obj = cls.__dataclass_fields__[field_name]
995 # Update the field default to None (overriding any parent class default)
996 field_obj.default = None
997 field_obj.default_factory = dataclasses.MISSING
999 # Also ensure the class attribute is None (should already be set, but double-check)
1000 setattr(cls, field_name, None)
1004def _inject_all_pending_fields():
1005 """Inject all accumulated fields at once."""
1006 for target_name, injection_data in _pending_injections.items():
1007 target_class = injection_data['target_class']
1008 configs = injection_data['configs_to_inject']
1010 if configs: # Only inject if there are configs to inject 1010 ↛ 1006line 1010 didn't jump to line 1006 because the condition on line 1010 was always true
1011 _inject_multiple_fields_into_dataclass(target_class, configs)
1013def _camel_to_snake(name: str) -> str:
1014 """Convert CamelCase to snake_case for field names."""
1015 s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
1016 return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
1018def _inject_multiple_fields_into_dataclass(target_class: Type, configs: List[Dict]) -> None:
1019 """Mathematical simplification: Batch field injection with direct dataclass recreation."""
1020 # Imports moved to top-level
1022 # Direct field reconstruction - guaranteed by dataclass contract
1023 existing_fields = [
1024 (f.name, f.type, field(default_factory=f.default_factory) if f.default_factory != MISSING
1025 else f.default if f.default != MISSING else f.type)
1026 for f in fields(target_class)
1027 ]
1029 # Mathematical simplification: Unified field construction with algebraic common factors
1030 def create_field_definition(config):
1031 """Create field definition with optional and inherit_as_none support."""
1032 field_type = config['config_class']
1033 is_optional = config.get('optional', False)
1034 is_ui_hidden = config.get('ui_hidden', False)
1036 # Algebraic simplification: factor out common default_value logic
1037 if is_optional: 1037 ↛ 1038line 1037 didn't jump to line 1038 because the condition on line 1037 was never true
1038 field_type = Union[field_type, type(None)]
1039 default_value = None
1040 else:
1041 # Both inherit_as_none and regular cases use same default factory
1042 # Add ui_hidden metadata to the field so UI layer can check it
1043 default_value = field(default_factory=field_type, metadata={'ui_hidden': is_ui_hidden})
1045 return (config['field_name'], field_type, default_value)
1047 all_fields = existing_fields + [create_field_definition(config) for config in configs]
1049 # Direct dataclass recreation - fail-loud
1050 new_class = make_dataclass(
1051 target_class.__name__,
1052 all_fields,
1053 bases=target_class.__bases__,
1054 frozen=target_class.__dataclass_params__.frozen
1055 )
1057 # CRITICAL: Preserve original module for proper imports in generated code
1058 # make_dataclass() sets __module__ to the caller's module (lazy_factory.py)
1059 # We need to set it to the target class's original module for correct import paths
1060 new_class.__module__ = target_class.__module__
1062 # Sibling inheritance is now handled by the dual-axis resolver system
1064 # Direct module replacement
1065 module = sys.modules[target_class.__module__]
1066 setattr(module, target_class.__name__, new_class)
1067 globals()[target_class.__name__] = new_class
1069 # Mathematical simplification: Extract common module assignment pattern
1070 def _register_lazy_class(lazy_class, class_name, module_name):
1071 """Register lazy class in both module and global namespace."""
1072 setattr(sys.modules[module_name], class_name, lazy_class)
1073 globals()[class_name] = lazy_class
1075 # Create lazy classes and recreate PipelineConfig inline
1076 for config in configs:
1077 lazy_class = LazyDataclassFactory.make_lazy_simple(
1078 base_class=config['config_class'],
1079 lazy_class_name=config['lazy_class_name']
1080 )
1081 _register_lazy_class(lazy_class, config['lazy_class_name'], config['config_class'].__module__)
1083 # Create lazy version of the updated global config itself with proper naming
1084 # Global configs must start with GLOBAL_CONFIG_PREFIX - fail-loud if not
1085 if not target_class.__name__.startswith(GLOBAL_CONFIG_PREFIX): 1085 ↛ 1086line 1085 didn't jump to line 1086 because the condition on line 1085 was never true
1086 raise ValueError(f"Target class '{target_class.__name__}' must start with '{GLOBAL_CONFIG_PREFIX}' prefix")
1088 # Remove global prefix (GlobalPipelineConfig → PipelineConfig)
1089 lazy_global_class_name = target_class.__name__[len(GLOBAL_CONFIG_PREFIX):]
1091 lazy_global_class = LazyDataclassFactory.make_lazy_simple(
1092 base_class=new_class,
1093 lazy_class_name=lazy_global_class_name
1094 )
1096 # Use extracted helper for consistent registration
1097 _register_lazy_class(lazy_global_class, lazy_global_class_name, target_class.__module__)
1103def auto_create_decorator(global_config_class):
1104 """
1105 Decorator that automatically creates:
1106 1. A field injection decorator for other configs to use
1107 2. A lazy version of the global config itself
1109 Global config classes must start with "Global" prefix.
1110 """
1111 # Validate naming convention
1112 if not global_config_class.__name__.startswith(GLOBAL_CONFIG_PREFIX): 1112 ↛ 1113line 1112 didn't jump to line 1113 because the condition on line 1112 was never true
1113 raise ValueError(f"Global config class '{global_config_class.__name__}' must start with '{GLOBAL_CONFIG_PREFIX}' prefix")
1115 decorator_name = _camel_to_snake(global_config_class.__name__)
1116 decorator = create_global_default_decorator(global_config_class)
1118 # Export decorator to module globals
1119 module = sys.modules[global_config_class.__module__]
1120 setattr(module, decorator_name, decorator)
1122 # Lazy global config will be created after field injection
1124 return global_config_class