Coverage for src/objectstate/lazy_factory.py: 81%
539 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""Generic lazy dataclass factory using flexible resolution."""
3# Standard library imports
4import copy
5import dataclasses
6import inspect
7import logging
8import re
9import sys
10from abc import ABC
11from contextlib import contextmanager
12from dataclasses import MISSING, dataclass, field, fields, is_dataclass, make_dataclass
13from functools import lru_cache
14from typing import Any, Callable, Dict, List, Optional, Tuple, Type, TypeVar, Union, get_type_hints
16from python_introspect import (
17 make_optional,
18 register_type_resolver,
19 resolve_annotated,
20 resolve_optional,
21)
23from objectstate.ui_visibility import mark_ui_hidden_config
25# Note: dual_axis_resolver_recursive and lazy_placeholder imports kept inline to avoid circular imports
28# Type registry for lazy dataclass to base class mapping
29_lazy_type_registry: Dict[Type, Type] = {}
31# Reverse registry for base class to lazy dataclass mapping (for O(1) lookup)
32_base_to_lazy_registry: Dict[Type, Type] = {}
35def _resolved_dataclass_annotations(dataclass_type: type) -> Dict[str, object]:
36 """Resolve declarations even while a class decorator precedes module binding."""
38 return get_type_hints(
39 dataclass_type,
40 localns={dataclass_type.__name__: dataclass_type},
41 include_extras=True,
42 )
45def _recreated_dataclass_namespace(
46 declared_type: type,
47 recreated_name: str,
48) -> dict[str, object]:
49 """Return declaration-owned metadata for a recreated dataclass.
51 The module and qualified name are in the namespace so the class has them
52 while it is being created: a metaclass that registers classes then sees the
53 declaration the class was rebuilt from, not ``abc`` or this module.
54 """
56 namespace: dict[str, object] = {
57 "__doc__": declared_type.__doc__,
58 "__module__": declared_type.__module__,
59 }
60 if recreated_name == declared_type.__name__:
61 namespace["__qualname__"] = declared_type.__qualname__
62 return namespace
65def _restore_recreated_dataclass_metadata(
66 recreated_type: type,
67 declared_type: type,
68) -> type:
69 """Restore runtime metadata that class creation may replace or remove."""
71 recreated_type.__module__ = declared_type.__module__
72 recreated_type.__doc__ = declared_type.__doc__
73 if recreated_type.__name__ == declared_type.__name__:
74 recreated_type.__qualname__ = declared_type.__qualname__
76 declared_first_line = declared_type.__dict__.get("__firstlineno__", MISSING)
77 if declared_first_line is not MISSING:
78 try:
79 setattr(recreated_type, "__firstlineno__", declared_first_line)
80 except (AttributeError, TypeError):
81 pass
82 return recreated_type
85# =============================================================================
86# UNIFIED NONE-FORCING: Single path for both base and lazy classes
87# Replaces the old 3-stage approach (pre-process setattr, post-process Field patch)
88# =============================================================================
90def get_inherited_field_names(cls: Type) -> set:
91 """
92 Get names of fields inherited from parent dataclasses (not defined in cls itself).
94 A field is "inherited" if it exists in a parent's __dataclass_fields__ but
95 is NOT in this class's own __annotations__ (i.e., not redefined here).
96 """
97 # Get all field names from parent dataclasses
98 parent_fields = set()
99 for base in cls.__mro__[1:]: # Skip cls itself
100 if dataclasses.is_dataclass(base):
101 parent_fields.update(base.__dataclass_fields__.keys())
103 # cls's OWN annotations (not inherited). Python 3.14 (PEP 649) no longer
104 # stores them in cls.__dict__['__annotations__'].
105 own_defined = set(inspect.get_annotations(cls, eval_str=False))
107 return parent_fields - own_defined
110def _inherited_default_metadata(field_definition) -> dict:
111 """Preserve one concrete dataclass field's standalone fallback values."""
113 metadata = dict(field_definition.metadata)
114 metadata.setdefault(
115 "_inherited_default",
116 (
117 field_definition.default
118 if field_definition.default is not MISSING
119 else MISSING
120 ),
121 )
122 metadata.setdefault(
123 "_inherited_default_factory",
124 field_definition.default_factory,
125 )
126 return metadata
129def rebuild_with_none_defaults(
130 cls: Type,
131 field_names_to_none: Optional[set] = None,
132 new_name: Optional[str] = None
133) -> Type:
134 """
135 Rebuild a dataclass via make_dataclass with None defaults for specified fields.
137 This is the UNIFIED approach for both base classes (inherit_as_none) and lazy classes.
138 Instead of patching Field objects after @dataclass, we rebuild with correct defaults.
140 Args:
141 cls: The dataclass to rebuild
142 field_names_to_none: Fields that should have default=None.
143 If None, ALL fields get default=None (for lazy classes).
144 new_name: Optional new class name (for lazy classes)
146 Returns:
147 A new class with the same fields but modified defaults
148 """
149 if not dataclasses.is_dataclass(cls):
150 raise ValueError(f"{cls} is not a dataclass")
152 if field_names_to_none is None:
153 # All fields get None (for lazy classes)
154 field_names_to_none = {f.name for f in fields(cls)}
156 annotations = _resolved_dataclass_annotations(cls)
158 # Build field definitions
159 field_defs = []
160 for f in fields(cls):
161 declared_type = annotations.get(f.name, f.type)
162 if f.name in field_names_to_none:
163 # Force None default, but preserve original default in metadata for fallback
164 # This allows standalone usage to fall back to parent's static default
165 field_defs.append(
166 (
167 f.name,
168 make_optional(declared_type),
169 field(
170 default=None,
171 metadata=_inherited_default_metadata(f),
172 ),
173 )
174 )
175 else:
176 # Preserve original field (copy to avoid sharing)
177 field_defs.append((f.name, declared_type, copy.copy(f)))
179 # Collect non-dunder attributes to preserve (methods, class vars, etc.)
180 recreated_name = new_name or cls.__name__
181 namespace = _recreated_dataclass_namespace(cls, recreated_name)
182 for key, value in cls.__dict__.items():
183 if key.startswith('__') and key.endswith('__'):
184 # Skip most dunders (make_dataclass will generate them)
185 # BUT preserve lifecycle/registration declarations.
186 if key not in {"__post_init__", "__registry_key__"}:
187 continue
188 if key == '__dataclass_fields__':
189 continue # Will be regenerated
190 namespace[key] = value
192 # Keep original bases for isinstance() to work
193 bases = cls.__bases__
195 # Check if any base is a frozen dataclass - if so, new class must also be frozen
196 is_frozen = any(
197 dataclasses.is_dataclass(b) and b.__dataclass_fields__ and
198 getattr(b, '__dataclass_params__', None) and b.__dataclass_params__.frozen
199 for b in cls.__mro__[1:]
200 )
202 # Get the original metaclass to preserve it
203 orig_metaclass = type(cls)
205 # Create new class
206 new_cls = make_dataclass(
207 recreated_name,
208 fields=field_defs,
209 bases=bases,
210 namespace=namespace,
211 frozen=is_frozen,
212 )
214 # make_dataclass takes no metaclass parameter: it derives one from the
215 # bases. When the declared class names a metaclass its bases do not carry,
216 # create the class once more with that metaclass. @dataclass is NOT
217 # re-applied, since make_dataclass already did that.
218 if type(new_cls) is not orig_metaclass:
219 new_cls = orig_metaclass(
220 new_cls.__name__,
221 new_cls.__bases__,
222 {**new_cls.__dict__, **_recreated_dataclass_namespace(cls, recreated_name)},
223 )
225 return _restore_recreated_dataclass_metadata(new_cls, cls)
228def replace_raw(instance, **changes):
229 """
230 Replace dataclass fields while preserving raw None values.
232 Unlike dataclasses.replace(), this function uses object.__getattribute__
233 to get field values, preventing lazy resolution from being triggered.
234 This is critical for lazy dataclasses where None means "inherit from parent"
235 and must not be resolved during copy operations.
237 Args:
238 instance: The dataclass instance to copy
239 **changes: Field values to override
241 Returns:
242 A new instance with raw values preserved (not resolved)
243 """
244 if not is_dataclass(instance):
245 raise TypeError(f"replace_raw() should be called on dataclass instances, got {type(instance)}")
247 # Get all field values using object.__getattribute__ to avoid lazy resolution
248 field_values = {}
249 for f in fields(instance):
250 if f.name in changes:
251 field_values[f.name] = changes[f.name]
252 else:
253 # Use object.__getattribute__ to get raw value (bypass lazy __getattribute__)
254 field_values[f.name] = object.__getattribute__(instance, f.name)
256 # Create new instance with raw values
257 return type(instance)(**field_values)
261def retain_only_authored_paths(value, authored_field_paths) -> Any:
262 """Copy a lazy dataclass, marking only ``authored_field_paths`` as explicitly set.
264 An ObjectState rebuild constructs every nested lazy dataclass from all its
265 flattened fields, so every constructor keyword reads as explicit. Clean
266 source generation needs the authored paths ObjectState already tracks
267 (its signature-diff fields); this applies them to the lazy metadata without
268 changing any field value.
269 """
271 from objectstate.field_access import DottedFieldPath
273 paths = tuple(DottedFieldPath(path) for path in authored_field_paths)
274 return _copy_with_only_authored_paths(value, DottedFieldPath(""), paths)
277def _copy_with_only_authored_paths(value, prefix, authored_field_paths):
278 from objectstate.field_access import DataclassFieldAccess
280 field_values = {}
281 for field_definition in fields(value):
282 field_value = DataclassFieldAccess.raw_value(value, field_definition.name)
283 if isinstance(field_value, LazyDataclass):
284 field_value = _copy_with_only_authored_paths(
285 field_value,
286 prefix.child(field_definition.name),
287 authored_field_paths,
288 )
289 field_values[field_definition.name] = field_value
291 projected = type(value)(**field_values)
292 object.__setattr__(
293 projected,
294 "_explicitly_set_fields",
295 {
296 field_definition.name
297 for field_definition in fields(projected)
298 if prefix.child(field_definition.name).contains_any(authored_field_paths)
299 },
300 )
301 return projected
303# ContextEventCoordinator removed - replaced with contextvars-based context system
308def register_lazy_type_mapping(lazy_type: Type, base_type: Type) -> None:
309 """Register mapping between lazy dataclass type and its base type."""
310 if not isinstance(lazy_type, type) or not isinstance(base_type, type):
311 raise TypeError("Lazy type mappings require two types.")
312 existing_base = _lazy_type_registry.get(lazy_type)
313 if existing_base is not None and existing_base is not base_type:
314 raise ValueError(
315 f"{lazy_type.__name__} is already registered for "
316 f"{existing_base.__name__}, not {base_type.__name__}."
317 )
318 existing_lazy = _base_to_lazy_registry.get(base_type)
319 if existing_lazy is not None and existing_lazy is not lazy_type:
320 raise ValueError(
321 f"{base_type.__name__} already owns lazy type "
322 f"{existing_lazy.__name__}, not {lazy_type.__name__}."
323 )
324 _lazy_type_registry[lazy_type] = base_type
325 _base_to_lazy_registry[base_type] = lazy_type
328def get_base_type_for_lazy(lazy_type: Type) -> Optional[Type]:
329 """Get the base type for a lazy dataclass type."""
330 return _lazy_type_registry.get(lazy_type)
333register_type_resolver(get_base_type_for_lazy)
336def is_lazy_dataclass(obj_or_type) -> bool:
337 """
338 Check if an object or type is a lazy dataclass.
340 ANTI-DUCK-TYPING: Uses isinstance() check against LazyDataclass base class
341 instead of hasattr() attribute sniffing.
343 Works with both instances and types, and naturally handles Optional types
344 without unwrapping.
346 Args:
347 obj_or_type: Either a dataclass instance or a dataclass type
349 Returns:
350 True if the object/type is a lazy dataclass
352 Examples:
353 >>> is_lazy_dataclass(PipelineConfig) # True (type check)
354 >>> is_lazy_dataclass(GlobalPipelineConfig) # False
355 >>> is_lazy_dataclass(pipeline_config_instance) # True (instance check)
356 >>> is_lazy_dataclass(LazyPathPlanningConfig) # True
357 >>> is_lazy_dataclass(PathPlanningConfig) # False
359 # Works with Optional without unwrapping!
360 >>> config: Optional[PipelineConfig] = PipelineConfig()
361 >>> is_lazy_dataclass(config) # True - checks the instance, not the type annotation
362 """
363 if isinstance(obj_or_type, type):
364 # Type check: is it a subclass of LazyDataclass?
365 return issubclass(obj_or_type, LazyDataclass)
366 else:
367 # Instance check: is it an instance of LazyDataclass?
368 return isinstance(obj_or_type, LazyDataclass)
370logger = logging.getLogger(__name__)
373# =============================================================================
374# GENERIC SCOPE RULE: Virtual base class for global configs using __instancecheck__
375# This allows isinstance() checks without actual inheritance, so lazy versions don't inherit it
376# =============================================================================
379_GLOBAL_CONFIG_TYPES: set[type] = set()
382def mark_global_config_type(config_type: Type) -> Type:
383 """Register config_type as an ObjectState global-config authority."""
384 if not isinstance(config_type, type) or not is_dataclass(config_type):
385 raise TypeError("Global config authorities must be dataclass types.")
386 _GLOBAL_CONFIG_TYPES.add(config_type)
387 return config_type
390class GlobalConfigMeta(type):
391 """
392 Metaclass that makes isinstance(obj, GlobalConfigBase) use the global config registry.
394 This enables type-safe isinstance checks without inheritance:
395 if isinstance(config, GlobalConfigBase): # Returns True for GlobalPipelineConfig
396 # Returns False for PipelineConfig (lazy version)
397 """
398 def __instancecheck__(cls, instance):
399 return is_global_config_type(type(instance))
402class GlobalConfigBase(metaclass=GlobalConfigMeta):
403 """
404 Virtual base class for all global config types.
406 Uses custom metaclass to check the global config registry instead of actual inheritance.
407 This prevents lazy versions (PipelineConfig) from being considered global configs.
409 Usage:
410 if isinstance(config, GlobalConfigBase): # Generic, works for any global config
412 Instead of:
413 if isinstance(config, GlobalPipelineConfig): # Hardcoded, breaks extensibility
414 """
415 pass
418class LazyResolutionDataclass(ABC):
419 """Nominal root for dataclasses whose ``None`` fields resolve by context."""
422class LazyDataclass(LazyResolutionDataclass):
423 """
424 Base class for all lazy dataclasses created by LazyDataclassFactory.
426 This enables isinstance() checks without duck typing or unwrapping:
427 isinstance(config, LazyDataclass) # Works!
428 isinstance(optional_config, LazyDataclass) # Works even for Optional!
430 All lazy dataclasses inherit from this, regardless of naming convention:
431 - PipelineConfig (lazy version of GlobalPipelineConfig)
432 - LazyPathPlanningConfig
433 - LazyPartitionFilterConfig
434 - etc.
436 ANTI-DUCK-TYPING: Use isinstance(obj, LazyDataclass) instead of hasattr() checks.
437 """
438 pass
441def has_lazy_resolution(obj_or_type: object) -> bool:
442 """Return whether a value or declared type owns lazy field resolution."""
444 candidate = resolve_optional(obj_or_type)
445 if isinstance(candidate, type):
446 return issubclass(candidate, LazyResolutionDataclass)
447 return isinstance(candidate, LazyResolutionDataclass)
450def is_global_config_type(config_type: Type) -> bool:
451 """
452 Check if a config type is a global config (marked by @auto_create_decorator).
454 GENERIC SCOPE RULE: Use this instead of hardcoding class name checks like:
455 if config_class == GlobalPipelineConfig:
457 Instead use:
458 if is_global_config_type(config_class):
460 Args:
461 config_type: The config class to check
463 Returns:
464 True if the type is marked as a global config, False otherwise
465 """
466 return config_type in _GLOBAL_CONFIG_TYPES
469def is_global_config_instance(config_instance: Any) -> bool:
470 """
471 Check if a config instance is an instance of a global config class.
473 GENERIC SCOPE RULE: Use this instead of hardcoding isinstance checks like:
474 if isinstance(config, GlobalPipelineConfig):
476 Instead use:
477 if is_global_config_instance(config):
479 Or use the virtual base class:
480 if isinstance(config, GlobalConfigBase):
482 Args:
483 config_instance: The config instance to check
485 Returns:
486 True if the instance is of a global config type, False otherwise
487 """
488 return is_global_config_type(type(config_instance))
491def get_lazy_type_for_base(base_type: Type) -> Optional[Type]:
492 """Get the lazy type for a base dataclass type."""
493 return _base_to_lazy_registry.get(base_type)
496def _normalized_config_type(annotation: object) -> Optional[type]:
497 """Return the nominal config owner declared by one field annotation."""
499 candidate = resolve_optional(resolve_annotated(annotation))
500 if not isinstance(candidate, type):
501 return None
502 return get_base_type_for_lazy(candidate) or candidate
505@lru_cache(maxsize=None)
506def _declared_config_field_names(
507 owner_type: type,
508 config_type: type,
509) -> Tuple[str, ...]:
510 """Find fields whose resolved declaration owns ``config_type``."""
512 if not is_dataclass(owner_type):
513 raise TypeError(f"{owner_type!r} is not a dataclass type.")
514 config_base = get_base_type_for_lazy(config_type) or config_type
515 annotations = _resolved_dataclass_annotations(owner_type)
516 return tuple(
517 field_definition.name
518 for field_definition in fields(owner_type)
519 if _normalized_config_type(
520 annotations.get(field_definition.name, field_definition.type)
521 )
522 is config_base
523 )
526def _single_declared_config_field(
527 owner_type: type,
528 config_type: type,
529 *,
530 required: bool,
531) -> Optional[str]:
532 """Select one annotation-owned config field and reject ambiguity."""
534 field_names = _declared_config_field_names(owner_type, config_type)
535 if len(field_names) == 1:
536 return field_names[0]
537 if not field_names and not required:
538 return None
539 config_base = get_base_type_for_lazy(config_type) or config_type
540 if not field_names:
541 raise TypeError(
542 f"{owner_type.__name__} has no field declared as "
543 f"{config_base.__name__}."
544 )
545 raise TypeError(
546 f"{owner_type.__name__} declares multiple {config_base.__name__} "
547 f"fields: {', '.join(field_names)}."
548 )
551# =============================================================================
552# Constants for lazy configuration system - simplified from class to module-level
553MATERIALIZATION_DEFAULTS_PATH = "materialization_defaults"
554RESOLVE_FIELD_VALUE_METHOD = "_resolve_field_value"
555GET_ATTRIBUTE_METHOD = "__getattribute__"
556TO_BASE_CONFIG_METHOD = "to_base_config"
557FROM_CONFIG_METHOD = "from_config"
558WITH_DEFAULTS_METHOD = "with_defaults"
559WITH_OVERRIDES_METHOD = "with_overrides"
560LAZY_FIELD_DEBUG_TEMPLATE = "LAZY FIELD CREATION: {field_name} - original={original_type}, has_default={has_default}, final={final_type}"
562LAZY_CLASS_NAME_PREFIX = "Lazy"
564# Legacy helper functions removed - new context system handles all resolution
567# Functional fallback strategies
568def _get_raw_field_value(obj: Any, field_name: str) -> Any:
569 """
570 Get raw field value bypassing lazy property getters to prevent infinite recursion.
572 Uses object.__getattribute__() to access stored values directly without triggering
573 lazy resolution, which would create circular dependencies in the resolution chain.
575 Args:
576 obj: Object to get field from
577 field_name: Name of field to access
579 Returns:
580 Raw field value or None if field doesn't exist
582 Raises:
583 AttributeError: If field doesn't exist (fail-loud behavior)
584 """
585 try:
586 return object.__getattribute__(obj, field_name)
587 except AttributeError:
588 return None
591def bind_lazy_resolution_to_class(cls: Type) -> None:
592 """
593 Add lazy __getattribute__ to an existing class.
595 This enables concrete classes (like PartitionFilterConfig stored in
596 GlobalPipelineConfig) to resolve None values via MRO without
597 changing their static defaults.
599 Args:
600 cls: The class to add lazy resolution to
601 """
602 if has_lazy_resolution(cls):
603 return
605 lazy_getattribute = LazyMethodBindings.create_getattribute()
606 cls.__getattribute__ = lazy_getattribute
607 LazyResolutionDataclass.register(cls)
610@dataclass(frozen=True)
611class LazyMethodBindings:
612 """Declarative method bindings for lazy dataclasses."""
614 @staticmethod
615 def create_resolver() -> Callable[[Any, str], Any]:
616 """Create the method that resolves one field in the active context."""
617 from objectstate.dual_axis_resolver import resolve_with_provenance
619 def _resolve_field_value(self, field_name: str) -> Any:
620 return resolve_with_provenance(type(self), field_name).value
622 return _resolve_field_value
624 @staticmethod
625 def create_getattribute() -> Callable[[Any, str], Any]:
626 """Create lazy __getattribute__: a None field reads its value from the active context."""
627 from objectstate.context_manager import context_layer_stack
628 from objectstate.dual_axis_resolver import declared_default, resolve_in_stack
630 def __getattribute__(self: Any, name: str) -> Any:
631 value = object.__getattribute__(self, name)
632 # Fast path: non-None values never need resolution
633 if value is not None:
634 return value
635 # PERFORMANCE: Use pre-computed frozenset (O(1) lookup) instead of
636 # rebuilding {f.name for f in fields(self.__class__)} (O(n)) on every access.
637 # Use object.__getattribute__ for __class__ to avoid recursion.
638 # Lazily populate if not yet set (e.g. class created outside _create_lazy_dataclass_unified).
639 _cls = object.__getattribute__(self, '__class__')
640 _fns = _cls.__dict__.get('_field_names_set')
641 if _fns is None:
642 _ft = fields(_cls)
643 _fns = frozenset(f.name for f in _ft)
644 _cls._field_names_set = _fns
645 _cls._fields_by_name = {f.name: f for f in _ft}
646 if name not in _fns:
647 return value
649 stack = context_layer_stack.get()
650 if not stack.layers:
651 return declared_default(_cls, name)[0]
652 resolved_value = resolve_in_stack(stack, _cls, name).value
653 if resolved_value is not None:
654 return resolved_value
656 # A nested dataclass field that nothing supplies reads as an empty instance.
657 field_obj = _cls.__dict__['_fields_by_name'].get(name)
658 if field_obj and is_dataclass(field_obj.type):
659 return field_obj.type()
660 return None
662 return __getattribute__
664 @staticmethod
665 def create_to_base_config(base_class: Type) -> Callable[[Any], Any]:
666 """Create base config converter method."""
667 def to_base_config(self):
668 # CRITICAL FIX: Use object.__getattribute__ to preserve raw None values
669 # getattr() triggers lazy resolution, converting None to static defaults
670 # None values must be preserved for dual-axis inheritance to work correctly
671 #
672 # Context: to_base_config() is called DURING config_context() setup (line 124 in context_manager.py)
673 # If we use getattr() here, it triggers resolution BEFORE the context is fully set up,
674 # causing resolution to use the wrong/stale context and losing the GlobalPipelineConfig base.
675 # We must extract raw None values here, let config_context() merge them into the hierarchy,
676 # and THEN resolution happens later with the properly built context.
677 field_values = {f.name: object.__getattribute__(self, f.name) for f in fields(self)}
678 return base_class(**field_values)
679 return to_base_config
681 @staticmethod
682 def create_from_config(base_class: Type) -> classmethod:
683 """Create one generic concrete projection and composition constructor."""
685 def from_config(cls, *configs, inherited=None):
686 if len(configs) == 1 and isinstance(configs[0], base_class):
687 config = configs[0]
688 if isinstance(config, LazyDataclass):
689 raise TypeError(
690 f"{cls.__name__}.from_config requires concrete "
691 f"{base_class.__name__}, got {type(config).__name__}."
692 )
693 if inherited is not None and (
694 isinstance(inherited, LazyDataclass)
695 or not isinstance(inherited, base_class)
696 ):
697 raise TypeError(
698 f"{cls.__name__}.from_config inherited value must be "
699 f"{base_class.__name__}, got {type(inherited).__name__}."
700 )
702 values = {}
703 for field_definition in fields(base_class):
704 if not field_definition.init:
705 continue
706 value = object.__getattribute__(config, field_definition.name)
707 if inherited is not None and value == object.__getattribute__(
708 inherited,
709 field_definition.name,
710 ):
711 continue
712 values[field_definition.name] = value
713 return cls(**values)
715 if inherited is not None:
716 raise TypeError(
717 f"{cls.__name__}.from_config accepts inherited only when "
718 f"projecting one {base_class.__name__}."
719 )
720 values = {}
721 for config in configs:
722 if isinstance(config, LazyDataclass):
723 raise TypeError(
724 f"{cls.__name__}.from_config requires concrete dataclass "
725 f"values, got {type(config).__name__}."
726 )
727 lazy_type = get_lazy_type_for_base(type(config))
728 if lazy_type is None:
729 raise TypeError(
730 f"{cls.__name__}.from_config has no registered lazy type "
731 f"for {type(config).__name__}."
732 )
733 field_name = _single_declared_config_field(
734 cls,
735 type(config),
736 required=True,
737 )
738 if field_name in values:
739 raise ValueError(
740 f"{cls.__name__}.from_config received duplicate config "
741 f"values for {field_name!r}."
742 )
743 values[field_name] = lazy_type.from_config(config)
744 return cls(**values)
746 return classmethod(from_config)
748 @staticmethod
749 def create_class_methods() -> Dict[str, Any]:
750 """Create class-level utility methods."""
751 return {
752 WITH_DEFAULTS_METHOD: classmethod(lambda cls: cls()),
753 WITH_OVERRIDES_METHOD: classmethod(lambda cls, **kwargs: cls(**kwargs))
754 }
757class LazyDataclassFactory:
758 """Generic factory for creating lazy dataclasses with flexible resolution."""
760 @staticmethod
761 def registered_type_pairs() -> tuple[tuple[type, type], ...]:
762 """Project current lazy/public declarations without constructing values."""
763 return tuple(_lazy_type_registry.items())
769 @staticmethod
770 def _introspect_dataclass_fields(
771 base_class: Type,
772 debug_template: str,
773 ) -> List[Tuple[str, Type, None]]:
774 """
775 Introspect dataclass fields for lazy loading.
777 Converts nested dataclass fields to lazy equivalents and makes fields Optional
778 if they lack defaults. Complex logic handles type unwrapping and lazy nesting.
779 """
780 base_fields = fields(base_class)
781 annotations = _resolved_dataclass_annotations(base_class)
782 lazy_field_definitions = []
784 for field_definition in base_fields:
785 # Check if field has default value or factory
786 has_default = (
787 field_definition.default is not MISSING
788 or field_definition.default_factory is not MISSING
789 )
791 # Check if field type is a dataclass that should be made lazy
792 field_type = annotations.get(
793 field_definition.name,
794 field_definition.type,
795 )
796 lazy_nested_type = None # Track if we created a lazy nested type
797 nested_type = resolve_annotated(field_type)
798 nested_is_registered_lazy = (
799 isinstance(nested_type, type)
800 and (
801 get_base_type_for_lazy(nested_type) is not None
802 or get_lazy_type_for_base(nested_type) is not None
803 or has_lazy_resolution(nested_type)
804 )
805 )
806 if is_dataclass(nested_type) and nested_is_registered_lazy:
807 if get_base_type_for_lazy(nested_type) is not None:
808 lazy_nested_type = nested_type
809 else:
810 lazy_nested_type = get_lazy_type_for_base(nested_type)
811 if lazy_nested_type is None:
812 lazy_nested_type = LazyDataclassFactory.make_lazy_simple(nested_type)
813 field_type = lazy_nested_type
814 logger.debug(f"Created lazy class for {field_definition.name}: {nested_type} -> {lazy_nested_type}")
816 # CRITICAL FIX: For lazy configs, nested dataclass fields should use default_factory
817 # to provide lazy instances (e.g., LazyPathPlanningConfig), not None.
818 # This allows getattr(pipeline_config, 'path_planning_config') to return an instance.
819 # Non-dataclass fields still default to None for placeholder inheritance.
820 # CRITICAL: Always preserve metadata from original field (e.g., ui_hidden flag)
821 if lazy_nested_type is not None:
822 final_field_type = field_type
823 # Nested dataclass field: use default_factory so accessing returns an instance
824 # This matches AbstractStep pattern: napari_streaming_config = LazyNapariStreamingConfig()
825 field_def = (
826 field_definition.name,
827 final_field_type,
828 dataclasses.field(
829 default_factory=lazy_nested_type,
830 metadata=field_definition.metadata,
831 ),
832 )
833 else:
834 final_field_type = make_optional(field_type)
835 # CRITICAL FIX: For lazy configs, ALL non-dataclass fields should default to None
836 # This enables proper inheritance from parent configs and placeholder styling
837 field_def = (
838 field_definition.name,
839 final_field_type,
840 dataclasses.field(
841 default=None,
842 metadata=_inherited_default_metadata(field_definition),
843 ),
844 )
846 lazy_field_definitions.append(field_def)
848 # Debug logging with provided template (reduced to DEBUG level to reduce log pollution)
849 logger.debug(debug_template.format(
850 field_name=field_definition.name,
851 original_type=field_definition.type,
852 has_default=has_default,
853 final_type=final_field_type
854 ))
856 return lazy_field_definitions
858 @staticmethod
859 def _create_lazy_dataclass_unified(
860 base_class: Type,
861 lazy_class_name: str | None,
862 debug_template: str,
863 ) -> Type:
864 """
865 Create lazy dataclass with declarative configuration.
867 Core factory method that creates lazy dataclass with introspected fields,
868 binds resolution methods, and registers type mappings. Complex orchestration
869 of field analysis, method binding, and class creation.
870 """
871 if not is_dataclass(base_class):
872 raise ValueError(f"{base_class} must be a dataclass")
874 registered_type = get_lazy_type_for_base(base_class)
875 if registered_type is not None:
876 if lazy_class_name not in (None, registered_type.__name__):
877 raise ValueError(
878 f"{base_class.__name__} already owns lazy type "
879 f"{registered_type.__name__}; cannot also create "
880 f"{lazy_class_name}."
881 )
882 return registered_type
883 if lazy_class_name is None:
884 lazy_class_name = f"{LAZY_CLASS_NAME_PREFIX}{base_class.__name__}"
886 # Create lazy dataclass with introspected fields
887 # CRITICAL FIX: Avoid inheriting from classes with custom metaclasses to prevent descriptor conflicts
888 # Exception: InheritAsNoneMeta is safe to inherit from as it only modifies field defaults
889 # Classes processed by the global-config decorator own lazy resolution
890 # nominally and are safe even when they use a custom metaclass.
891 base_metaclass = type(base_class)
892 has_unsafe_metaclass = (
893 (hasattr(base_class, '__metaclass__') or base_metaclass != type) and
894 not has_lazy_resolution(base_class)
895 )
897 # Determine inheritance: always include LazyDataclass, optionally include base_class
898 if has_unsafe_metaclass:
899 # Base class has unsafe custom metaclass - don't inherit, just copy interface
900 logger.debug(
901 "Lazy factory: %s has custom metaclass %s, avoiding inheritance",
902 base_class.__name__,
903 base_metaclass.__name__,
904 )
905 bases = (LazyDataclass,) # Only inherit from LazyDataclass
906 else:
907 # Safe to inherit from regular dataclass
908 bases = (base_class, LazyDataclass) # Inherit from both
910 # Check if base_class is frozen - must match to avoid inheritance error
911 # "cannot inherit frozen dataclass from a non-frozen one" and vice versa
912 base_is_frozen = (
913 is_dataclass(base_class) and
914 hasattr(base_class, '__dataclass_params__') and
915 base_class.__dataclass_params__.frozen
916 )
918 # Single make_dataclass call - no duplication
919 lazy_class = make_dataclass(
920 lazy_class_name,
921 LazyDataclassFactory._introspect_dataclass_fields(
922 base_class,
923 debug_template,
924 ),
925 bases=bases,
926 namespace=_recreated_dataclass_namespace(
927 base_class,
928 lazy_class_name,
929 ),
930 frozen=base_is_frozen # Match base class frozen status
931 )
933 # PERFORMANCE: Cache field names set and field-by-name dict per class.
934 # These are derived from dataclasses.fields() which is immutable after class creation.
935 # Avoids O(n) set creation on every __getattribute__ call and O(n) linear scan
936 # on every field lookup.
937 from dataclasses import fields as _dc_fields
938 _field_tuple = _dc_fields(lazy_class)
939 lazy_class._field_names_set = frozenset(f.name for f in _field_tuple)
940 lazy_class._fields_by_name = {f.name: f for f in _field_tuple}
942 # Add constructor parameter tracking to detect user-set fields
943 original_init = lazy_class.__init__
944 def __init_with_tracking__(self, **kwargs):
945 # Track which fields were explicitly passed to constructor
946 object.__setattr__(self, '_explicitly_set_fields', set(kwargs.keys()))
947 original_init(self, **kwargs)
949 lazy_class.__init__ = __init_with_tracking__
951 # Bind methods declaratively - inline single-use method
952 method_bindings = {
953 RESOLVE_FIELD_VALUE_METHOD: LazyMethodBindings.create_resolver(),
954 GET_ATTRIBUTE_METHOD: LazyMethodBindings.create_getattribute(),
955 TO_BASE_CONFIG_METHOD: LazyMethodBindings.create_to_base_config(base_class),
956 FROM_CONFIG_METHOD: LazyMethodBindings.create_from_config(base_class),
957 **LazyMethodBindings.create_class_methods()
958 }
959 for method_name, method_impl in method_bindings.items():
960 setattr(lazy_class, method_name, method_impl)
962 _restore_recreated_dataclass_metadata(lazy_class, base_class)
964 # Automatically register the lazy dataclass with the type registry
965 register_lazy_type_mapping(lazy_class, base_class)
967 return lazy_class
973 @staticmethod
974 def make_lazy_simple(
975 base_class: Type,
976 lazy_class_name: str | None = None
977 ) -> Type:
978 """
979 Create lazy dataclass using new contextvars system.
981 SIMPLIFIED: No complex hierarchy providers or field path detection needed.
982 Uses new contextvars system for all resolution.
984 Args:
985 base_class: Base dataclass to make lazy
986 lazy_class_name: Name for the lazy class. When omitted the factory
987 names it, or returns the lazy class the base already owns.
989 Returns:
990 Generated lazy dataclass with contextvars-based resolution
991 """
992 return LazyDataclassFactory._create_lazy_dataclass_unified(
993 base_class=base_class,
994 lazy_class_name=lazy_class_name,
995 debug_template=f"Simple contextvars resolution for {base_class.__name__}",
996 )
998 # All legacy methods removed - use make_lazy_simple() for all use cases
1001# =============================================================================
1002# Constructor Patching for Code Execution
1003# =============================================================================
1005@contextmanager
1006def patch_lazy_constructors(types: Optional[List[Type]] = None):
1007 """
1008 Context manager that patches lazy dataclass constructors to preserve None vs concrete distinction.
1010 This is critical for code editors that use exec() to create dataclass instances.
1011 Without patching, lazy dataclasses would resolve None values to concrete defaults
1012 during construction, making it impossible to distinguish between explicitly set
1013 values and inherited values.
1015 The patched constructor sets fields provided in kwargs and otherwise uses the
1016 dataclass defaults/default_factory (or None if none exist). This preserves the
1017 None vs concrete distinction while still instantiating nested lazy configs.
1019 Args:
1020 types: Optional list of lazy types to patch. If None, uses every type
1021 in the authoritative lazy-to-base registry.
1023 Usage:
1024 # Patch during code execution
1025 with patch_lazy_constructors():
1026 exec(code_string, namespace)
1027 # Lazy dataclasses created during exec() will preserve None values
1029 Example:
1030 # Without patching:
1031 LazyZarrConfig(compression='gzip') # All unspecified fields resolve to defaults
1033 # With patching:
1034 with patch_lazy_constructors():
1035 LazyZarrConfig(compression='gzip') # Only compression is set, rest are None
1036 """
1037 lazy_types = (
1038 list(_lazy_type_registry)
1039 if types is None
1040 else list(types)
1041 )
1043 if not lazy_types:
1044 # No types to patch - just yield
1045 yield
1046 return
1048 # Store original constructors
1049 original_constructors: Dict[Type, callable] = {}
1051 # Patch all lazy types
1052 for lazy_type in lazy_types:
1053 # Store original constructor
1054 original_constructors[lazy_type] = lazy_type.__init__
1056 # Create patched constructor that uses raw values
1057 def create_patched_init(dataclass_type, original_init):
1058 def patched_init(self, **kwargs):
1059 # Materialize the raw constructor payload without reading fields
1060 # through lazy ``__getattribute__``. The nominal constructor must
1061 # still run: it owns dataclass ``__post_init__`` validation and any
1062 # other declaration-level invariants.
1063 raw_values = {}
1064 for field_definition in dataclasses.fields(dataclass_type):
1065 if field_definition.name in kwargs:
1066 value = kwargs[field_definition.name]
1067 elif field_definition.default_factory is not dataclasses.MISSING: # type: ignore
1068 value = field_definition.default_factory()
1069 elif field_definition.default is not dataclasses.MISSING:
1070 value = field_definition.default
1071 else:
1072 value = None
1074 raw_values[field_definition.name] = value
1076 original_init(self, **raw_values)
1078 # The generated lazy constructor tracks every received keyword.
1079 # Restore the authored subset because the remaining raw values were
1080 # supplied only to preserve inheritance during reconstruction.
1081 object.__setattr__(self, '_explicitly_set_fields', set(kwargs.keys()))
1083 return patched_init
1085 # Apply the patch
1086 lazy_type.__init__ = create_patched_init(
1087 lazy_type,
1088 original_constructors[lazy_type],
1089 )
1091 try:
1092 yield
1093 finally:
1094 # Restore original constructors
1095 for lazy_type, original_init in original_constructors.items():
1096 lazy_type.__init__ = original_init
1099# ContextProvider infrastructure removed - was dead code feeding broken frame.f_locals manipulation
1104def resolve_lazy_configurations_for_serialization(data: Any) -> Any:
1105 """
1106 Recursively resolve lazy dataclass instances to concrete values for serialization.
1108 CRITICAL: This function must be called WITHIN a config_context() block!
1109 The context provides the hierarchy for lazy resolution.
1111 How it works:
1112 1. For lazy dataclasses: Access fields with getattr() to trigger resolution
1113 2. The lazy __getattribute__ uses the active config_context() to resolve None values
1114 3. Convert resolved values to base config for pickling
1116 Example (from README.md):
1117 with config_context(orchestrator.pipeline_config):
1118 # Lazy resolution happens here via context
1119 resolved_steps = resolve_lazy_configurations_for_serialization(steps)
1120 """
1121 # Check if this is a lazy dataclass
1122 base_type = get_base_type_for_lazy(type(data))
1123 if base_type is not None:
1124 # This is a lazy dataclass - resolve fields using getattr() within the active context
1125 # getattr() triggers lazy __getattribute__ which uses config_context() for resolution
1126 resolved_fields = {}
1127 for f in fields(data):
1128 # CRITICAL: Use getattr() to trigger lazy resolution via context
1129 # The active config_context() provides the hierarchy for resolution
1130 resolved_value = getattr(data, f.name)
1131 resolved_fields[f.name] = resolved_value
1133 # Create base config instance with resolved values
1134 resolved_data = base_type(**resolved_fields)
1135 else:
1136 # Not a lazy dataclass
1137 resolved_data = data
1139 # Recursively process nested structures based on type
1140 if is_dataclass(resolved_data) and not isinstance(resolved_data, type):
1141 # Process dataclass fields recursively
1142 logger.debug(f"Resolving fields for {type(resolved_data).__name__}: {[f.name for f in fields(resolved_data)]}")
1143 resolved_fields = {}
1144 for f in fields(resolved_data):
1145 field_value = getattr(resolved_data, f.name)
1146 logger.debug(f"Resolving {type(resolved_data).__name__}.{f.name} = {type(field_value).__name__}")
1147 resolved_fields[f.name] = resolve_lazy_configurations_for_serialization(field_value)
1148 cls = type(resolved_data)
1150 # Optional hook: allow dataclasses with custom constructors to control rebuild.
1151 rebuild = getattr(cls, "__objectstate_rebuild__", None)
1152 if callable(rebuild):
1153 return rebuild(**resolved_fields)
1155 # Default: try normal construction, then fall back to field-wise construction.
1156 try:
1157 return cls(**resolved_fields)
1158 except TypeError:
1159 obj = cls.__new__(cls)
1160 for k, v in resolved_fields.items():
1161 object.__setattr__(obj, k, v)
1162 return obj
1164 elif isinstance(resolved_data, dict):
1165 # Process dictionary values recursively
1166 return {
1167 key: resolve_lazy_configurations_for_serialization(value)
1168 for key, value in resolved_data.items()
1169 }
1171 elif isinstance(resolved_data, (list, tuple)):
1172 # Process sequence elements recursively
1173 resolved_items = [resolve_lazy_configurations_for_serialization(item) for item in resolved_data]
1174 return type(resolved_data)(resolved_items)
1176 else:
1177 # Primitive type or unknown structure - return as-is
1178 return resolved_data
1181# Generic dataclass editing with configurable value preservation
1182T = TypeVar('T')
1185def create_dataclass_for_editing(dataclass_type: Type[T], source_config: Any, preserve_values: bool = False, context_provider: Optional[Callable[[Any], None]] = None) -> T:
1186 """Create dataclass for editing with configurable value preservation."""
1187 if not is_dataclass(dataclass_type):
1188 raise ValueError(f"{dataclass_type} must be a dataclass")
1190 # Set up context if provider is given (e.g., thread-local storage)
1191 if context_provider:
1192 context_provider(source_config)
1194 field_values = {
1195 f.name: (getattr(source_config, f.name) if preserve_values
1196 else f.type() if is_dataclass(f.type) and has_lazy_resolution(f.type)
1197 else None)
1198 for f in fields(dataclass_type)
1199 }
1201 return dataclass_type(**field_values)
1207# Declarative Global Config Field Injection System
1208# Moved inline imports to top-level
1210# Naming configuration
1211GLOBAL_CONFIG_PREFIX = "Global"
1213@dataclass(frozen=True)
1214class ConfigInjection:
1215 """An injected config owns its annotation, default and dataclass field tuple."""
1217 config_class: Type
1218 field_name: str
1219 nullable: bool = False
1220 ui_hidden: bool = False
1222 def field_tuple(self) -> Tuple[str, object, dataclasses.Field]:
1223 metadata = {'ui_hidden': self.ui_hidden}
1224 if self.nullable:
1225 return self.field_name, Optional[self.config_class], field(default=None, metadata=metadata)
1226 return self.field_name, self.config_class, field(default_factory=self.config_class, metadata=metadata)
1229_pending_injections: Dict[Type, List[ConfigInjection]] = {}
1231# Preview label registry: Type -> label string
1232# Used by UI to auto-discover which configs should appear in list item previews
1233PREVIEW_LABEL_REGISTRY: Dict[Type, str] = {}
1235# Field abbreviations registry: Type -> {field_name: abbreviation}
1236# Used by UI to display compact field names in list item previews
1237FIELD_ABBREVIATIONS_REGISTRY: Dict[Type, Dict[str, str]] = {}
1239# Class attribute holding a config class's own group abbreviation. The class
1240# is the only owner; get_group_abbreviation reads it on demand.
1241GROUP_ABBREVIATION_ATTRIBUTE = "_abbreviation"
1243# Always viewable fields registry: Type -> List[field_name]
1244# Used by UI to auto-discover which fields should always be shown in list item previews
1245# regardless of the specific widget's PREVIEW_FIELD_CONFIGS.
1246# This allows config types to declare their own preview fields declaratively.
1247ALWAYS_VIEWABLE_FIELDS_REGISTRY: Dict[Type, List[str]] = {}
1250class abbreviation:
1251 """
1252 Decorator/marker for abbreviations in config classes.
1254 Can be used as:
1255 1. Class decorator: @abbreviation('wfc') to set class abbreviation
1256 2. In Annotated types: Annotated[Type, abbreviation('pf')] for field abbreviation
1258 Examples:
1259 @abbreviation('wfc')
1260 @global_pipeline_config
1261 @dataclass
1262 class PartitionFilterConfig:
1263 partition_filter: Annotated[Optional[int], abbreviation('pf')] = None
1264 partition_filter_mode: Annotated[
1265 type[PartitionFilterMode], abbreviation('pfm')
1266 ] = IncludePartitions
1268 Args:
1269 name: The abbreviation string
1271 Returns:
1272 When used as class decorator: the class, declaring its group abbreviation
1273 When used in Annotated: a marker object storing the abbreviation
1274 """
1276 def __init__(self, name: str):
1277 self.name = name
1279 def __call__(self, cls: Type) -> Type:
1280 """Class decorator usage: @abbreviation('wfc')"""
1281 declare_group_abbreviation(cls, self.name)
1282 return cls
1284 def __repr__(self) -> str:
1285 return f"abbreviation({self.name!r})"
1288def _extract_abbreviations_from_annotations(cls: Type) -> Dict[str, str]:
1289 """
1290 Extract field abbreviations from Annotated type hints.
1292 Scans class annotations for fields using Annotated[..., abbreviation('...')]
1293 and returns a dict mapping field names to abbreviations.
1295 Walks the MRO from most-specific to least-specific, so child class
1296 annotations override parent class annotations (provenance wins).
1298 Args:
1299 cls: The class to scan
1301 Returns:
1302 Dict mapping field names to abbreviation strings
1303 """
1304 from typing import get_origin, get_args
1306 field_abbreviations = {}
1308 # Walk MRO from most-specific to least-specific
1309 # This ensures child class annotations override parent class annotations
1310 for klass in cls.__mro__:
1311 if klass is object:
1312 continue
1314 # Get this class's own annotations (not inherited)
1315 annotations = inspect.get_annotations(klass, eval_str=False)
1317 for field_name, field_type in annotations.items():
1318 # Check if this is an Annotated type
1319 origin = get_origin(field_type)
1320 if origin is not None:
1321 # Get the args - first is the actual type, rest are metadata
1322 args = get_args(field_type)
1323 if len(args) > 1:
1324 # Check metadata for abbreviation markers
1325 for metadata in args[1:]:
1326 if isinstance(metadata, abbreviation):
1327 # Child class definitions override parent class
1328 # because we walk MRO most-specific first
1329 field_abbreviations[field_name] = metadata.name
1330 break
1332 return field_abbreviations
1335def declare_group_abbreviation(config_type: type, name: str) -> None:
1336 """Declare a config class's own group abbreviation.
1338 A lazy class has no abbreviation of its own: the declaration lands on its
1339 base class. A second, different declaration for one class raises.
1340 """
1341 declaring_type = get_base_type_for_lazy(config_type) or config_type
1342 declared = vars(declaring_type).get(GROUP_ABBREVIATION_ATTRIBUTE)
1343 if declared is not None and declared != name:
1344 raise ValueError(
1345 f"{declaring_type.__qualname__} already declares abbreviation "
1346 f"{declared!r}; cannot also declare {name!r}."
1347 )
1348 setattr(declaring_type, GROUP_ABBREVIATION_ATTRIBUTE, name)
1351def get_group_abbreviation(config_type: str | type) -> str:
1352 """Read a config class's group abbreviation from its declaration.
1354 A lazy class answers with its base class's declaration. A class with no
1355 declaration of its own answers with its nearest declaring ancestor, then
1356 with the prefix of its class name. A type name string answers with its
1357 prefix.
1358 """
1359 if isinstance(config_type, str):
1360 return config_type.split('_')[0] if config_type else "root"
1362 declaring_type = get_base_type_for_lazy(config_type) or config_type
1363 for ancestor in declaring_type.__mro__:
1364 declared = vars(ancestor).get(GROUP_ABBREVIATION_ATTRIBUTE)
1365 if declared is not None:
1366 return declared
1367 return config_type.__name__.split('_')[0]
1370def create_global_default_decorator(target_config_class: Type):
1371 """
1372 Create a decorator factory for a specific global config class.
1374 The decorator accumulates all decorations, then injects all fields at once
1375 when the module finishes loading. Also creates lazy versions of all decorated configs.
1376 """
1377 _pending_injections.setdefault(target_config_class, [])
1379 def global_default_decorator(cls=None, *, optional: bool = False, inherit_as_none: bool = True, ui_hidden: bool = False, preview_label: Optional[str] = None, abbreviation: Optional[str] = None, field_abbreviations: Optional[Dict[str, str]] = None, always_viewable_fields: Optional[List[str]] = None):
1380 """
1381 Decorator that can be used with or without parameters.
1383 Args:
1384 cls: The class being decorated (when used without parentheses)
1385 optional: Whether to wrap the field type with Optional (default: False)
1386 inherit_as_none: Whether to set all inherited fields to None by default (default: True)
1387 ui_hidden: Whether to hide from UI (apply decorator but don't inject into global config) (default: False)
1388 preview_label: Short label for list item previews (e.g., "NAP", "FIJI", "MAT"). If set,
1389 config will appear in preview when enabled. Registered in PREVIEW_LABEL_REGISTRY.
1390 abbreviation: Short abbreviation for config class name used in grouped previews (e.g., "pp" for PathPlanningConfig).
1391 Declared on the class; see declare_group_abbreviation.
1392 field_abbreviations: Dict mapping field names to abbreviations for compact display.
1393 E.g., {'partition_filter': 'pf', 'num_workers': 'W'}. Registered in
1394 FIELD_ABBREVIATIONS_REGISTRY.
1395 always_viewable_fields: List of field names that should always be shown in list item previews
1396 for this config type. E.g., ['enabled', 'persistent']. Registered in ALWAYS_VIEWABLE_FIELDS_REGISTRY.
1397 """
1398 def decorator(actual_cls):
1399 # UNIFIED NONE-FORCING: Single make_dataclass rebuild instead of old 3-stage approach
1400 if inherit_as_none:
1401 # Rebuild class with None defaults for inherited fields
1402 # This replaces the old pre-process setattr + post-process Field patching
1403 inherited_fields = get_inherited_field_names(actual_cls)
1404 if inherited_fields:
1405 actual_cls = rebuild_with_none_defaults(actual_cls, inherited_fields)
1407 field_name = _camel_to_snake(actual_cls.__name__)
1409 # Mark class with typed ui_hidden metadata for UI layer to check
1410 # This allows the config to remain in the context (for lazy resolution)
1411 # while being hidden from UI rendering
1412 if ui_hidden:
1413 mark_ui_hidden_config(actual_cls)
1415 # Register preview label for UI list item previews
1416 # Allows ABC to auto-discover which configs should appear in preview
1417 if preview_label is not None:
1418 PREVIEW_LABEL_REGISTRY[actual_cls] = preview_label
1420 # The class owns its group abbreviation; @abbreviation writes the
1421 # same declaration, so the two decorators may come in either order.
1422 if abbreviation is not None:
1423 declare_group_abbreviation(actual_cls, abbreviation)
1425 # Register field abbreviations for compact preview display
1426 # Priority: explicit parameter > @abbreviation in Annotated types
1427 detected_field_abbrs = _extract_abbreviations_from_annotations(actual_cls)
1428 final_field_abbrs = {**detected_field_abbrs, **(field_abbreviations or {})}
1429 logger.debug(f"🔍 Field abbrs detected={detected_field_abbrs}, explicit={field_abbreviations}, final={final_field_abbrs}")
1430 if final_field_abbrs:
1431 FIELD_ABBREVIATIONS_REGISTRY[actual_cls] = final_field_abbrs
1433 # Register always viewable fields for this config type
1434 # These fields will always be shown in list item previews regardless of widget config
1435 if always_viewable_fields:
1436 ALWAYS_VIEWABLE_FIELDS_REGISTRY[actual_cls] = always_viewable_fields
1437 logger.debug(f"🔍 Registered always_viewable_fields for {actual_cls.__name__}: {always_viewable_fields}")
1439 # Check if class is abstract (has unimplemented abstract methods)
1440 # Abstract classes should NEVER be injected into GlobalPipelineConfig
1441 # because they can't be instantiated
1442 # NOTE: We need to check if the class ITSELF is abstract, not just if it inherits from ABC
1443 # Concrete subclasses of abstract classes should still be injected
1444 # We check for __abstractmethods__ attribute which exists even before @dataclass runs
1445 # (it's set by ABCMeta when the class is created)
1446 is_abstract = hasattr(actual_cls, '__abstractmethods__') and len(actual_cls.__abstractmethods__) > 0
1448 # Add to pending injections for field injection
1449 # Skip injection for abstract classes (they can't be instantiated)
1450 # For concrete classes: inject even if ui_hidden (needed for lazy resolution context)
1451 if not is_abstract:
1452 _pending_injections[target_config_class].append(
1453 ConfigInjection(actual_cls, field_name, nullable=optional, ui_hidden=ui_hidden))
1455 # Immediately create lazy version of this config (not dependent on injection)
1458 # Declare contextual resolution on the concrete type before deriving
1459 # its lazy form. The nominal relationship, not a marker attribute,
1460 # is the capability contract used by the factory and its consumers.
1461 bind_lazy_resolution_to_class(actual_cls)
1463 lazy_class = LazyDataclassFactory.make_lazy_simple(actual_cls)
1465 # Export lazy class to config module immediately
1466 config_module = sys.modules[actual_cls.__module__]
1467 setattr(config_module, lazy_class.__name__, lazy_class)
1469 # Copy metadata to lazy class for UI compatibility
1470 if ui_hidden:
1471 mark_ui_hidden_config(lazy_class)
1472 if preview_label is not None:
1473 PREVIEW_LABEL_REGISTRY[lazy_class] = preview_label
1474 # Copy field abbreviations to lazy class (from explicit param or detected from Annotated)
1475 field_abbr_to_copy = field_abbreviations or detected_field_abbrs
1476 if field_abbr_to_copy:
1477 FIELD_ABBREVIATIONS_REGISTRY[lazy_class] = field_abbr_to_copy
1479 # Copy always_viewable_fields to lazy class
1480 if always_viewable_fields:
1481 ALWAYS_VIEWABLE_FIELDS_REGISTRY[lazy_class] = always_viewable_fields
1483 # Note: No Stage 3 post-processing needed!
1484 # - Base class: rebuilt via rebuild_with_none_defaults() above
1485 # - Lazy class: _introspect_dataclass_fields() already sets None defaults
1487 return actual_cls
1489 # Handle both @decorator and @decorator() usage
1490 if cls is None:
1491 # Called with parentheses: @decorator(optional=True)
1492 return decorator
1493 else:
1494 # Called without parentheses: @decorator
1495 return decorator(cls)
1497 return global_default_decorator
1500def _inject_all_pending_fields():
1501 """Inject all accumulated fields at once."""
1502 for target_class, configs in _pending_injections.items():
1503 if configs: # Only inject if there are configs to inject
1504 _inject_multiple_fields_into_dataclass(target_class, configs)
1506def _camel_to_snake(name: str) -> str:
1507 """Convert CamelCase to snake_case for field names."""
1508 s1 = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name)
1509 return re.sub('([a-z0-9])([A-Z])', r'\1_\2', s1).lower()
1511def _inject_multiple_fields_into_dataclass(target_class: Type, configs: List[ConfigInjection]) -> None:
1512 """Mathematical simplification: Batch field injection with direct dataclass recreation."""
1513 # Imports moved to top-level
1515 annotations = _resolved_dataclass_annotations(target_class)
1516 existing_fields = [
1517 (f.name, annotations.get(f.name, f.type), copy.copy(f))
1518 for f in fields(target_class)
1519 ]
1521 all_fields = existing_fields + [config.field_tuple() for config in configs]
1523 # Direct dataclass recreation - fail-loud
1524 new_class = make_dataclass(
1525 target_class.__name__,
1526 all_fields,
1527 bases=target_class.__bases__,
1528 namespace=_recreated_dataclass_namespace(
1529 target_class,
1530 target_class.__name__,
1531 ),
1532 frozen=target_class.__dataclass_params__.frozen
1533 )
1535 _restore_recreated_dataclass_metadata(new_class, target_class)
1537 # Preserve global-config registration across dataclass recreation.
1538 # Registration is set by @auto_create_decorator but lost when make_dataclass creates a new class.
1539 if is_global_config_type(target_class):
1540 mark_global_config_type(new_class)
1541 # Sibling inheritance is now handled by the dual-axis resolver system
1543 # Direct module replacement
1544 module = sys.modules[target_class.__module__]
1545 setattr(module, target_class.__name__, new_class)
1546 globals()[target_class.__name__] = new_class
1548 # Mathematical simplification: Extract common module assignment pattern
1549 def _register_lazy_class(lazy_class, class_name, module_name):
1550 """Register lazy class in both module and global namespace."""
1551 setattr(sys.modules[module_name], class_name, lazy_class)
1552 globals()[class_name] = lazy_class
1554 # Create lazy classes and recreate PipelineConfig inline
1555 for config in configs:
1556 lazy_class = LazyDataclassFactory.make_lazy_simple(config.config_class)
1557 _register_lazy_class(lazy_class, lazy_class.__name__, config.config_class.__module__)
1559 # Create lazy version of the updated global config itself with proper naming
1560 # Global configs must start with GLOBAL_CONFIG_PREFIX - fail-loud if not
1561 if not target_class.__name__.startswith(GLOBAL_CONFIG_PREFIX):
1562 raise ValueError(f"Target class '{target_class.__name__}' must start with '{GLOBAL_CONFIG_PREFIX}' prefix")
1564 # Remove global prefix (GlobalPipelineConfig → PipelineConfig)
1565 lazy_global_class_name = target_class.__name__[len(GLOBAL_CONFIG_PREFIX):]
1567 lazy_global_class = LazyDataclassFactory.make_lazy_simple(
1568 base_class=new_class,
1569 lazy_class_name=lazy_global_class_name
1570 )
1572 # Use extracted helper for consistent registration
1573 _register_lazy_class(lazy_global_class, lazy_global_class_name, target_class.__module__)
1579def auto_create_decorator(global_config_class):
1580 """
1581 Decorator that automatically creates:
1582 1. A field injection decorator for other configs to use
1583 2. A lazy version of the global config itself
1585 Global config classes must start with "Global" prefix.
1586 """
1587 # Validate naming convention
1588 if not global_config_class.__name__.startswith(GLOBAL_CONFIG_PREFIX):
1589 raise ValueError(f"Global config class '{global_config_class.__name__}' must start with '{GLOBAL_CONFIG_PREFIX}' prefix")
1591 # Mark this class as a global config for isinstance checks via GlobalConfigBase.
1592 mark_global_config_type(global_config_class)
1594 decorator_name = _camel_to_snake(global_config_class.__name__)
1595 decorator = create_global_default_decorator(global_config_class)
1597 # Export decorator to module globals
1598 module = sys.modules[global_config_class.__module__]
1599 setattr(module, decorator_name, decorator)
1601 # Lazy global config will be created after field injection
1603 return global_config_class