Coverage for src/hieraconf/dual_axis_resolver.py: 16%
190 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-02 21:44 +0000
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-02 21:44 +0000
1"""
2Generic dual-axis resolver for lazy configuration inheritance.
4This module provides the core inheritance resolution logic as a pure function,
5supporting both context hierarchy (X-axis) and sibling inheritance (Y-axis).
7The resolver is completely generic and has no application-specific dependencies.
8"""
10import logging
11from typing import Any, Dict, Type, Optional
12from dataclasses import is_dataclass
14logger = logging.getLogger(__name__)
17def _has_concrete_field_override(source_class, field_name: str) -> bool:
18 """
19 Check if a class has a concrete field override (not None).
21 This determines inheritance design based on static class definition:
22 - Concrete default (not None) = never inherits
23 - None default = always inherits (inherit_as_none design)
24 """
25 # CRITICAL FIX: Check class attribute directly, not dataclass field default
26 # The @global_pipeline_config decorator modifies field defaults to None
27 # but the class attribute retains the original concrete value
28 if hasattr(source_class, field_name):
29 class_attr_value = getattr(source_class, field_name)
30 has_override = class_attr_value is not None
31 return has_override
32 return False
35# Priority functions removed - MRO-based resolution is sufficient
38def resolve_field_inheritance_old(
39 obj,
40 field_name: str,
41 available_configs: Dict[str, Any]
42) -> Any:
43 """
44 Pure function for cross-dataclass inheritance resolution.
46 This replaces the complex RecursiveContextualResolver with explicit parameter passing.
48 Args:
49 obj: The object requesting field resolution
50 field_name: Name of the field to resolve
51 available_configs: Dict mapping config type names to config instances
52 e.g., {'GlobalPipelineConfig': global_config, 'StepConfig': step_config}
54 Returns:
55 Resolved field value or None if not found
57 Algorithm:
58 1. Check if obj has concrete value for field_name
59 2. Check Y-axis inheritance within obj's MRO for concrete values
60 3. Check related config types in available_configs for cross-dataclass inheritance
61 4. Return class defaults as final fallback
62 """
63 obj_type = type(obj)
65 # Step 1: Check concrete value in merged context for obj's type (HIGHEST PRIORITY)
66 # CRITICAL: Context values take absolute precedence over inheritance blocking
67 # The config_context() manager merges concrete values into available_configs
68 for config_name, config_instance in available_configs.items():
69 if type(config_instance) == obj_type:
70 try:
71 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
72 value = object.__getattribute__(config_instance, field_name)
73 if value is not None:
74 if field_name == 'well_filter':
75 logger.debug(f"🔍 CONTEXT: Found concrete value in merged context {obj_type.__name__}.{field_name}: {value}")
76 return value
77 except AttributeError:
78 # Field doesn't exist on this config type
79 continue
81 # Step 1b: Check concrete value on obj instance itself (fallback)
82 # Use object.__getattribute__ to avoid recursion with lazy __getattribute__
83 try:
84 value = object.__getattribute__(obj, field_name)
85 if value is not None:
86 if field_name == 'well_filter':
87 logger.debug(f"🔍 INSTANCE: Found concrete value on instance {obj_type.__name__}.{field_name}: {value}")
88 return value
89 except AttributeError:
90 # Field doesn't exist on the object
91 pass
93 # Step 2: FIELD-SPECIFIC INHERITANCE BLOCKING
94 # Check if this specific field has a concrete value in the exact same type
95 # Only block inheritance if the EXACT same type has a non-None value
96 for config_name, config_instance in available_configs.items():
97 if type(config_instance) == obj_type:
98 try:
99 field_value = object.__getattribute__(config_instance, field_name)
100 if field_value is not None:
101 # This exact type has a concrete value - use it, don't inherit
102 if field_name == 'well_filter':
103 logger.debug(f"🔍 FIELD-SPECIFIC BLOCKING: {obj_type.__name__}.{field_name} = {field_value} (concrete) - blocking inheritance")
104 return field_value
105 except AttributeError:
106 continue
108 # DEBUG: Log what we're trying to resolve
109 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
110 logger.debug(f"🔍 RESOLVING {obj_type.__name__}.{field_name} - checking context and inheritance")
111 logger.debug(f"🔍 AVAILABLE CONFIGS: {list(available_configs.keys())}")
113 # Step 3: Y-axis inheritance within obj's MRO
114 blocking_class = _find_blocking_class_in_mro(obj_type, field_name)
116 for parent_type in obj_type.__mro__[1:]:
117 if not is_dataclass(parent_type):
118 continue
120 # Check blocking logic
121 if blocking_class and parent_type != blocking_class:
122 continue
124 if blocking_class and parent_type == blocking_class:
125 # Check if blocking class has concrete value in available configs
126 for config_name, config_instance in available_configs.items():
127 if type(config_instance) == parent_type:
128 try:
129 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
130 value = object.__getattribute__(config_instance, field_name)
131 if value is None:
132 # Blocking class has None - inheritance blocked
133 break
134 else:
135 logger.debug(f"Inherited from blocking class {parent_type.__name__}: {value}")
136 return value
137 except AttributeError:
138 # Field doesn't exist on this config type
139 continue
140 break
142 # Normal inheritance - check for concrete values
143 for config_name, config_instance in available_configs.items():
144 if type(config_instance) == parent_type:
145 try:
146 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
147 value = object.__getattribute__(config_instance, field_name)
148 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
149 logger.debug(f"🔍 Y-AXIS INHERITANCE: {parent_type.__name__}.{field_name} = {value}")
150 if value is not None:
151 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
152 logger.debug(f"🔍 Y-AXIS INHERITANCE: FOUND {parent_type.__name__}.{field_name}: {value} (returning)")
153 logger.debug(f"Inherited from {parent_type.__name__}: {value}")
154 return value
155 except AttributeError:
156 # Field doesn't exist on this config type
157 continue
159 # Step 4: Cross-dataclass inheritance from related config types (MRO-based)
160 # NOTE: Inheritance blocking was already applied in Step 2, so this only runs for types without concrete overrides
161 # Uses pure MRO-based resolution - no custom priority functions needed
162 for config_name, config_instance in available_configs.items():
163 config_type = type(config_instance)
165 if _is_related_config_type(obj_type, config_type):
166 # Skip if this is the same type as the requesting object (avoid self-inheritance)
167 if config_type == obj_type:
168 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
169 logger.debug(f"🔍 CROSS-DATACLASS: Skipping self-inheritance from {config_type.__name__}")
170 continue
172 try:
173 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
174 value = object.__getattribute__(config_instance, field_name)
175 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
176 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__}.{field_name} = {value} (related config)")
177 if value is not None:
178 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
179 logger.debug(f"🔍 CROSS-DATACLASS: FOUND {config_type.__name__}.{field_name}: {value}")
180 logger.debug(f"Cross-dataclass inheritance from {config_type.__name__}: {value}")
181 return value
182 except AttributeError:
183 # Field doesn't exist on this config type
184 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
185 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__} has no field {field_name}")
186 continue
187 else:
188 if field_name in ['output_dir_suffix', 'sub_dir']:
189 logger.debug(f"🔍 CROSS-DATACLASS: {config_type.__name__} not related to {obj_type.__name__}")
191 # Step 4: Class defaults as final fallback
192 if blocking_class:
193 try:
194 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
195 class_default = object.__getattribute__(blocking_class, field_name)
196 if class_default is not None:
197 logger.debug(f"Using class default from blocking class {blocking_class.__name__}: {class_default}")
198 return class_default
199 except AttributeError:
200 # Field doesn't exist on blocking class
201 pass
203 logger.debug(f"No resolution found for {obj_type.__name__}.{field_name}")
204 return None
207def _is_related_config_type(obj_type: Type, config_type: Type) -> bool:
208 """
209 Check if config_type is related to obj_type for cross-dataclass inheritance.
211 CRITICAL FIX: Only allow inheritance from parent classes or sibling classes at the same level,
212 NOT from child classes. This prevents WellFilterConfig from inheriting from StepWellFilterConfig.
214 Args:
215 obj_type: The type requesting field resolution
216 config_type: The type being checked for relationship
218 Returns:
219 True if config_type should be considered for cross-dataclass inheritance
220 """
221 # CRITICAL: Only allow inheritance from parent classes (obj_type inherits from config_type)
222 # This prevents base classes from inheriting from their derived classes
223 if issubclass(obj_type, config_type):
224 return True
226 # Allow sibling inheritance only if they share a common parent but neither inherits from the other
227 # This allows StepMaterializationConfig to inherit from both StepWellFilterConfig and PathPlanningConfig
228 if not issubclass(config_type, obj_type): # config_type is NOT a child of obj_type
229 # Check if they share a common dataclass ancestor (excluding themselves)
230 obj_ancestors = set(cls for cls in obj_type.__mro__[1:] if is_dataclass(cls)) # Skip obj_type itself
231 config_ancestors = set(cls for cls in config_type.__mro__[1:] if is_dataclass(cls)) # Skip config_type itself
233 shared_ancestors = obj_ancestors & config_ancestors
234 if shared_ancestors:
235 return True
237 return False
240def resolve_field_inheritance(
241 obj,
242 field_name: str,
243 available_configs: Dict[str, Any]
244) -> Any:
245 """
246 Simplified MRO-based inheritance resolution.
248 ALGORITHM:
249 1. Check if obj has concrete value for field_name in context
250 2. Traverse obj's MRO from most to least specific
251 3. For each MRO class, check if there's a config instance in context with concrete (non-None) value
252 4. Return first concrete value found
254 Args:
255 obj: The object requesting field resolution
256 field_name: Name of the field to resolve
257 available_configs: Dict mapping config type names to config instances
259 Returns:
260 Resolved field value or None if not found
261 """
262 obj_type = type(obj)
264 # Step 1: Check if exact same type has concrete value in context
265 for config_name, config_instance in available_configs.items():
266 if type(config_instance) == obj_type:
267 try:
268 field_value = object.__getattribute__(config_instance, field_name)
269 if field_value is not None:
270 if field_name == 'well_filter':
271 logger.debug(f"🔍 CONCRETE VALUE: {obj_type.__name__}.{field_name} = {field_value}")
272 return field_value
273 except AttributeError:
274 continue
276 # Step 2: MRO-based inheritance - traverse MRO from most to least specific
277 # For each class in the MRO, check if there's a config instance in context with concrete value
278 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
279 logger.debug(f"🔍 MRO-INHERITANCE: Resolving {obj_type.__name__}.{field_name}")
280 logger.debug(f"🔍 MRO-INHERITANCE: MRO = {[cls.__name__ for cls in obj_type.__mro__]}")
282 for mro_class in obj_type.__mro__:
283 if not is_dataclass(mro_class):
284 continue
286 # Look for a config instance of this MRO class type in the available configs
287 for config_name, config_instance in available_configs.items():
288 if type(config_instance) == mro_class:
289 try:
290 value = object.__getattribute__(config_instance, field_name)
291 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
292 logger.debug(f"🔍 MRO-INHERITANCE: {mro_class.__name__}.{field_name} = {value}")
293 if value is not None:
294 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
295 logger.debug(f"🔍 MRO-INHERITANCE: FOUND {mro_class.__name__}.{field_name}: {value} (returning)")
296 return value
297 except AttributeError:
298 continue
300 # Step 3: Class defaults as final fallback
301 try:
302 class_default = object.__getattribute__(obj_type, field_name)
303 if class_default is not None:
304 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
305 logger.debug(f"🔍 CLASS-DEFAULT: {obj_type.__name__}.{field_name} = {class_default}")
306 return class_default
307 except AttributeError:
308 pass
310 if field_name in ['output_dir_suffix', 'sub_dir', 'well_filter']:
311 logger.debug(f"🔍 NO-RESOLUTION: {obj_type.__name__}.{field_name} = None")
312 return None
315# Utility functions for inheritance detection (kept from original resolver)
317def _has_concrete_field_override(config_class: Type, field_name: str) -> bool:
318 """
319 Check if a class has a concrete field override (not None).
321 This determines class-level inheritance blocking behavior based on static class definition.
322 Now checks the entire MRO chain to handle inherited fields properly.
323 """
324 try:
325 # Check the entire MRO chain for concrete field values
326 for cls in config_class.__mro__:
327 if hasattr(cls, field_name):
328 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
329 class_attr_value = object.__getattribute__(cls, field_name)
330 if class_attr_value is not None:
331 has_override = True
332 logger.debug(f"Class override check {config_class.__name__}.{field_name}: found concrete value {class_attr_value} in {cls.__name__}, has_override={has_override}")
333 return has_override
335 # No concrete value found in any class in the MRO
336 logger.debug(f"Class override check {config_class.__name__}.{field_name}: no concrete value in MRO, has_override=False")
337 return False
338 except AttributeError:
339 # Field doesn't exist on class
340 return False
343def _find_blocking_class_in_mro(base_type: Type, field_name: str) -> Optional[Type]:
344 """
345 Find the first class in MRO that has a concrete field override AND blocks inheritance from parent classes.
347 A class blocks inheritance only if:
348 1. It has a concrete field override
349 2. There are parent classes in the MRO that also have the same field
351 This prevents legitimate inheritance sources (like GlobalPipelineConfig) from being treated as blockers.
353 Returns:
354 The first class in MRO order that blocks inheritance, or None if no blocking class found.
355 """
356 for i, cls in enumerate(base_type.__mro__):
357 if not is_dataclass(cls):
358 continue
359 if _has_concrete_field_override(cls, field_name):
360 # Check if there are parent classes that also have this field
361 has_parent_with_field = False
362 for parent_cls in base_type.__mro__[i + 1:]:
363 if not is_dataclass(parent_cls):
364 continue
365 try:
366 # Use object.__getattribute__ to avoid triggering lazy __getattribute__ recursion
367 object.__getattribute__(parent_cls, field_name)
368 has_parent_with_field = True
369 break
370 except AttributeError:
371 # Field doesn't exist on this parent class
372 continue
374 if has_parent_with_field:
375 logger.debug(f"Found blocking class {cls.__name__} for {base_type.__name__}.{field_name} (blocks parent inheritance)")
376 return cls
377 else:
378 logger.debug(f"Class {cls.__name__} has concrete override but no parents with field - not blocking")
379 return None
382# All legacy functions removed - use resolve_field_inheritance() instead