Coverage for openhcs/config_framework/context_manager.py: 44.3%
208 statements
« prev ^ index » next coverage.py v7.10.7, created at 2025-10-01 18:33 +0000
1"""
2Generic contextvars-based context management system for lazy configuration.
4This module provides explicit context scoping using Python's contextvars to enable
5hierarchical configuration resolution without explicit parameter passing.
7Key features:
81. Explicit context scoping with config_context() manager
92. Config extraction from functions, dataclasses, and objects
103. Config merging for context hierarchy
114. Clean separation between UI windows and contexts
13Key components:
14- current_temp_global: ContextVar holding current merged global config
15- config_context(): Context manager for creating context scopes
16- extract_config_overrides(): Extract config values from any object type
17- merge_configs(): Merge overrides into base config
18"""
20import contextvars
21import dataclasses
22import inspect
23import logging
24from contextlib import contextmanager
25from typing import Any, Dict, Optional, Type, Union
26from dataclasses import fields, is_dataclass
28logger = logging.getLogger(__name__)
30# Core contextvar for current merged global config
31# This holds the current context state that resolution functions can access
32current_temp_global = contextvars.ContextVar('current_temp_global')
35def _merge_nested_dataclass(base, override):
36 """
37 Recursively merge nested dataclass fields.
39 For each field in override:
40 - If value is None: skip (don't override base)
41 - If value is dataclass: recursively merge with base's value
42 - Otherwise: use override value
44 Args:
45 base: Base dataclass instance
46 override: Override dataclass instance
48 Returns:
49 Merged dataclass instance
50 """
51 if not is_dataclass(base) or not is_dataclass(override): 51 ↛ 52line 51 didn't jump to line 52 because the condition on line 51 was never true
52 return override
54 merge_values = {}
55 for field_info in fields(override):
56 field_name = field_info.name
57 override_value = object.__getattribute__(override, field_name)
59 if override_value is None:
60 # None means "don't override" - keep base value
61 continue
62 elif is_dataclass(override_value): 62 ↛ 64line 62 didn't jump to line 64 because the condition on line 62 was never true
63 # Recursively merge nested dataclass
64 base_value = getattr(base, field_name, None)
65 if base_value is not None and is_dataclass(base_value):
66 merge_values[field_name] = _merge_nested_dataclass(base_value, override_value)
67 else:
68 merge_values[field_name] = override_value
69 else:
70 # Concrete value - use override
71 merge_values[field_name] = override_value
73 # Merge with base
74 if merge_values:
75 return dataclasses.replace(base, **merge_values)
76 else:
77 return base
80@contextmanager
81def config_context(obj):
82 """
83 Create new context scope with obj's matching fields merged into base config.
85 This is the universal context manager for all config context needs. It works by:
86 1. Finding fields that exist on both obj and the base config type
87 2. Using matching field values to create a temporary merged config
88 3. Setting that as the current context
90 Args:
91 obj: Object with config fields (pipeline_config, step, etc.)
93 Usage:
94 with config_context(orchestrator.pipeline_config): # Pipeline-level context
95 # ...
96 with config_context(step): # Step-level context
97 # ...
98 """
99 # Get current context as base for nested contexts, or fall back to base global config
100 current_context = get_current_temp_global()
101 base_config = current_context if current_context is not None else get_base_global_config()
103 # Find matching fields between obj and base config type
104 overrides = {}
105 if obj is not None: 105 ↛ 145line 105 didn't jump to line 145 because the condition on line 105 was always true
106 from openhcs.config_framework.config import get_base_config_type
108 base_config_type = get_base_config_type()
110 for field_info in fields(base_config_type):
111 field_name = field_info.name
112 expected_type = field_info.type
114 # Check if obj has this field
115 try:
116 # Use object.__getattribute__ to avoid triggering lazy resolution
117 if hasattr(obj, field_name):
118 value = object.__getattribute__(obj, field_name)
119 if value is not None:
120 # Check if value is compatible (handles lazy-to-base type mapping)
121 if _is_compatible_config_type(value, expected_type): 121 ↛ 110line 121 didn't jump to line 110 because the condition on line 121 was always true
122 # Convert lazy configs to base configs for context
123 if hasattr(value, 'to_base_config'): 123 ↛ 129line 123 didn't jump to line 129 because the condition on line 123 was always true
124 value = value.to_base_config()
126 # CRITICAL FIX: Recursively merge nested dataclass fields
127 # If this is a dataclass field, merge it with the base config's value
128 # instead of replacing wholesale
129 if is_dataclass(value): 129 ↛ 140line 129 didn't jump to line 140 because the condition on line 129 was always true
130 base_value = getattr(base_config, field_name, None)
131 if base_value is not None and is_dataclass(base_value): 131 ↛ 137line 131 didn't jump to line 137 because the condition on line 131 was always true
132 # Merge nested dataclass: base + overrides
133 merged_nested = _merge_nested_dataclass(base_value, value)
134 overrides[field_name] = merged_nested
135 else:
136 # No base value to merge with, use override as-is
137 overrides[field_name] = value
138 else:
139 # Non-dataclass field, use override as-is
140 overrides[field_name] = value
141 except AttributeError:
142 continue
144 # Create merged config if we have overrides
145 if overrides: 145 ↛ 153line 145 didn't jump to line 153 because the condition on line 145 was always true
146 try:
147 merged_config = dataclasses.replace(base_config, **overrides)
148 logger.debug(f"Creating config context with {len(overrides)} field overrides from {type(obj).__name__}")
149 except Exception as e:
150 logger.warning(f"Failed to merge config overrides from {type(obj).__name__}: {e}")
151 merged_config = base_config
152 else:
153 merged_config = base_config
154 logger.debug(f"Creating config context with no overrides from {type(obj).__name__}")
156 token = current_temp_global.set(merged_config)
157 try:
158 yield
159 finally:
160 current_temp_global.reset(token)
163# Removed: extract_config_overrides - no longer needed with field matching approach
166# UNUSED: Kept for compatibility but no longer used with field matching approach
167def extract_from_function_signature(func) -> Dict[str, Any]:
168 """
169 Get parameter defaults as config overrides.
171 This enables functions to provide config context through their parameter defaults.
172 Useful for step functions that want to specify their own config values.
174 Args:
175 func: Function to extract parameter defaults from
177 Returns:
178 Dict of parameter_name -> default_value for parameters with defaults
179 """
180 try:
181 sig = inspect.signature(func)
182 overrides = {}
184 for name, param in sig.parameters.items():
185 if param.default != inspect.Parameter.empty:
186 overrides[name] = param.default
188 logger.debug(f"Extracted {len(overrides)} overrides from function {func.__name__}")
189 return overrides
191 except (ValueError, TypeError) as e:
192 logger.debug(f"Could not extract signature from {func}: {e}")
193 return {}
196def extract_from_dataclass_fields(obj) -> Dict[str, Any]:
197 """
198 Get non-None fields as config overrides.
200 This extracts concrete values from dataclass instances, ignoring None values
201 which represent fields that should inherit from context.
203 Args:
204 obj: Dataclass instance to extract field values from
206 Returns:
207 Dict of field_name -> value for non-None fields
208 """
209 if not is_dataclass(obj):
210 return {}
212 overrides = {}
214 for field in fields(obj):
215 value = getattr(obj, field.name)
216 if value is not None:
217 overrides[field.name] = value
219 logger.debug(f"Extracted {len(overrides)} overrides from dataclass {type(obj).__name__}")
220 return overrides
223def extract_from_object_attributes(obj) -> Dict[str, Any]:
224 """
225 Extract config attributes from step/pipeline objects.
227 This handles orchestrators, steps, and other objects that have *_config attributes.
228 It flattens the config hierarchy into a single dict of field overrides.
230 Args:
231 obj: Object to extract config attributes from
233 Returns:
234 Dict of field_name -> value for all non-None config fields
235 """
236 overrides = {}
238 try:
239 for attr_name in dir(obj):
240 if attr_name.endswith('_config'):
241 attr_value = getattr(obj, attr_name)
242 if attr_value is not None and is_dataclass(attr_value):
243 # Extract all non-None fields from this config
244 config_overrides = extract_from_dataclass_fields(attr_value)
245 overrides.update(config_overrides)
247 logger.debug(f"Extracted {len(overrides)} overrides from object {type(obj).__name__}")
249 except Exception as e:
250 logger.debug(f"Error extracting from object {obj}: {e}")
252 return overrides
255def merge_configs(base, overrides: Dict[str, Any]):
256 """
257 Merge overrides into base config, creating new immutable instance.
259 This creates a new config instance with override values merged in,
260 preserving immutability of the original base config.
262 Args:
263 base: Base config instance (base config type)
264 overrides: Dict of field_name -> value to override
266 Returns:
267 New config instance with overrides applied
268 """
269 if not base or not overrides:
270 return base
272 try:
273 # Filter out None values - they should not override existing values
274 filtered_overrides = {k: v for k, v in overrides.items() if v is not None}
276 if not filtered_overrides:
277 return base
279 # Use dataclasses.replace to create new instance with overrides
280 merged = dataclasses.replace(base, **filtered_overrides)
282 logger.debug(f"Merged {len(filtered_overrides)} overrides into {type(base).__name__}")
283 return merged
285 except Exception as e:
286 logger.warning(f"Failed to merge configs: {e}")
287 return base
290def get_base_global_config():
291 """
292 Get the base global config (fallback when no context set).
294 This provides the global config that was set up with ensure_global_config_context(),
295 or a default if none was set. Used as the base for merging operations.
297 Returns:
298 Current global config instance or default instance of base config type
299 """
300 try:
301 from openhcs.config_framework.config import get_base_config_type
302 from openhcs.config_framework.global_config import get_current_global_config
304 base_config_type = get_base_config_type()
306 # First try to get the global config that was set up
307 current_global = get_current_global_config(base_config_type)
308 if current_global is not None: 308 ↛ 312line 308 didn't jump to line 312 because the condition on line 308 was always true
309 return current_global
311 # Fallback to default if none was set
312 return base_config_type()
313 except ImportError:
314 logger.warning("Could not get base config type")
315 return None
318def get_current_temp_global():
319 """
320 Get current context or None.
322 This is the primary interface for resolution functions to access
323 the current context. Returns None if no context is active.
325 Returns:
326 Current merged global config or None
327 """
328 return current_temp_global.get(None)
331def set_current_temp_global(config):
332 """
333 Set current context (for testing/debugging).
335 This is primarily for testing purposes. Normal code should use
336 config_context() manager instead.
338 Args:
339 config: Global config instance to set as current context
341 Returns:
342 Token for resetting the context
343 """
344 return current_temp_global.set(config)
347def clear_current_temp_global():
348 """
349 Clear current context (for testing/debugging).
351 This removes any active context, causing resolution to fall back
352 to default behavior.
353 """
354 try:
355 current_temp_global.set(None)
356 except LookupError:
357 pass # No context was set
360# Utility functions for debugging and introspection
362def get_context_info() -> Dict[str, Any]:
363 """
364 Get information about current context for debugging.
366 Returns:
367 Dict with context information including type, field count, etc.
368 """
369 current = get_current_temp_global()
370 if current is None:
371 return {"active": False}
373 return {
374 "active": True,
375 "type": type(current).__name__,
376 "field_count": len(fields(current)) if is_dataclass(current) else 0,
377 "non_none_fields": sum(1 for f in fields(current)
378 if getattr(current, f.name) is not None) if is_dataclass(current) else 0
379 }
382def extract_all_configs_from_context() -> Dict[str, Any]:
383 """
384 Extract all *_config attributes from current context.
386 This is used by the resolution system to get all available configs
387 for cross-dataclass inheritance resolution.
389 Returns:
390 Dict of config_name -> config_instance for all *_config attributes
391 """
392 current = get_current_temp_global()
393 if current is None:
394 return {}
396 return extract_all_configs(current)
399def extract_all_configs(context_obj) -> Dict[str, Any]:
400 """
401 Extract all config instances from a context object using type-driven approach.
403 This function leverages dataclass field type annotations to efficiently extract
404 config instances, avoiding string matching and runtime attribute scanning.
406 Args:
407 context_obj: Object to extract configs from (orchestrator, merged config, etc.)
409 Returns:
410 Dict mapping config type names to config instances
411 """
412 if context_obj is None: 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 return {}
415 configs = {}
417 # Include the context object itself if it's a dataclass
418 if is_dataclass(context_obj): 418 ↛ 422line 418 didn't jump to line 422 because the condition on line 418 was always true
419 configs[type(context_obj).__name__] = context_obj
421 # Type-driven extraction: Use dataclass field annotations to find config fields
422 if is_dataclass(type(context_obj)): 422 ↛ 449line 422 didn't jump to line 449 because the condition on line 422 was always true
423 for field_info in fields(type(context_obj)):
424 field_type = field_info.type
425 field_name = field_info.name
427 # Handle Optional[ConfigType] annotations
428 actual_type = _unwrap_optional_type(field_type)
430 # Only process fields that are dataclass types (config objects)
431 if is_dataclass(actual_type):
432 try:
433 field_value = getattr(context_obj, field_name)
434 if field_value is not None: 434 ↛ 423line 434 didn't jump to line 423 because the condition on line 434 was always true
435 # Use the actual instance type, not the annotation type
436 # This handles cases where field is annotated as base class but contains subclass
437 instance_type = type(field_value)
438 configs[instance_type.__name__] = field_value
440 logger.debug(f"Extracted config {instance_type.__name__} from field {field_name}")
442 except AttributeError:
443 # Field doesn't exist on instance (shouldn't happen with dataclasses)
444 logger.debug(f"Field {field_name} not found on {type(context_obj).__name__}")
445 continue
447 # For non-dataclass objects (orchestrators, etc.), extract dataclass attributes
448 else:
449 _extract_from_object_attributes_typed(context_obj, configs)
451 logger.debug(f"Extracted {len(configs)} configs: {list(configs.keys())}")
452 return configs
455def _unwrap_optional_type(field_type):
456 """
457 Unwrap Optional[T] and Union[T, None] types to get the actual type T.
459 This handles type annotations like Optional[ConfigType] -> ConfigType
460 """
461 # Handle typing.Optional and typing.Union
462 if hasattr(field_type, '__origin__'): 462 ↛ 463line 462 didn't jump to line 463 because the condition on line 462 was never true
463 if field_type.__origin__ is Union:
464 # Get non-None types from Union
465 non_none_types = [arg for arg in field_type.__args__ if arg is not type(None)]
466 if len(non_none_types) == 1:
467 return non_none_types[0]
469 return field_type
472def _extract_from_object_attributes_typed(obj, configs: Dict[str, Any]) -> None:
473 """
474 Type-safe extraction from object attributes for non-dataclass objects.
476 This is used for orchestrators and other objects that aren't dataclasses
477 but have config attributes. Uses type checking instead of string matching.
478 """
479 try:
480 # Get all attributes that are dataclass instances
481 for attr_name in dir(obj):
482 if attr_name.startswith('_'):
483 continue
485 try:
486 attr_value = getattr(obj, attr_name)
487 if attr_value is not None and is_dataclass(attr_value):
488 configs[type(attr_value).__name__] = attr_value
489 logger.debug(f"Extracted config {type(attr_value).__name__} from attribute {attr_name}")
491 except (AttributeError, TypeError):
492 # Skip attributes that can't be accessed or aren't relevant
493 continue
495 except Exception as e:
496 logger.debug(f"Error in typed attribute extraction: {e}")
499def _is_compatible_config_type(value, expected_type) -> bool:
500 """
501 Check if value is compatible with expected_type, handling lazy-to-base type mapping.
503 This handles cases where:
504 - value is LazyStepMaterializationConfig, expected_type is StepMaterializationConfig
505 - value is a subclass of the expected type
506 - value is exactly the expected type
507 """
508 value_type = type(value)
510 # Direct type match
511 if value_type == expected_type: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true
512 return True
514 # Check if value_type is a subclass of expected_type
515 try:
516 if issubclass(value_type, expected_type): 516 ↛ 523line 516 didn't jump to line 523 because the condition on line 516 was always true
517 return True
518 except TypeError:
519 # expected_type might not be a class (e.g., Union, Optional)
520 pass
522 # Check lazy-to-base type mapping
523 if hasattr(value, 'to_base_config'):
524 # This is a lazy config - check if its base type matches expected_type
525 from openhcs.config_framework.lazy_factory import _lazy_type_registry
526 base_type = _lazy_type_registry.get(value_type)
527 if base_type == expected_type:
528 return True
529 # Also check if base type is subclass of expected type
530 if base_type and issubclass(base_type, expected_type):
531 return True
533 return False