Coverage for src/objectstate/live_context_resolver.py: 91%
54 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""
2Live context resolution service for configuration framework.
4Provides cached resolution of config attributes using live values,
5avoiding redundant context building and resolution operations.
7This service is completely generic and UI-agnostic:
8- No knowledge of ParameterFormManager or any UI components
9- No knowledge of steps, orchestrators, or domain concepts
10- Only knows about dataclasses, context stacks, and attribute resolution
11- Caller is responsible for providing live context data
12"""
14import logging
15from dataclasses import is_dataclass
16from typing import Any
18from objectstate.context_manager import config_context
19from objectstate.lazy_factory import replace_raw
21logger = logging.getLogger(__name__)
24class _ResolvedAttr:
25 """One answer, held together with the objects it was computed from.
27 Holding the objects keeps them alive for as long as the answer is kept, and
28 the answer is found again by comparing those objects with ``is``.
29 """
31 __slots__ = ("config_obj", "context_stack", "value")
33 def __init__(self, config_obj: object, context_stack: tuple[object, ...], value: Any):
34 self.config_obj = config_obj
35 self.context_stack = context_stack
36 self.value = value
38 def computed_from(self, config_obj: object, context_stack: list) -> bool:
39 return (
40 self.config_obj is config_obj
41 and len(self.context_stack) == len(context_stack)
42 and all(kept is given for kept, given in zip(self.context_stack, context_stack))
43 )
46class LiveContextResolver:
47 """
48 Pure service for resolving config attributes with live values.
50 Keeps resolved values to avoid expensive context stack building + resolution.
51 The answers kept are those of one cache token. Each is stored with the config
52 object and the context objects it was computed from, so it is returned only
53 for those same objects. Answers of an earlier token are dropped when a new
54 token arrives.
56 Completely generic - works with any dataclasses and any context hierarchy.
57 UI layer is responsible for:
58 - Collecting live context from form managers
59 - Providing the current token
60 - Building the context stack
61 """
63 def __init__(self):
64 self._answers_token: int | None = None
65 self._answers_by_attr: dict[str, list[_ResolvedAttr]] = {}
67 def resolve_config_attr(
68 self,
69 config_obj: object,
70 attr_name: str,
71 context_stack: list,
72 live_context: dict[type, dict[str, Any]],
73 cache_token: int,
74 ) -> Any:
75 """
76 Resolve config attribute through context hierarchy with caching.
78 Completely generic - no knowledge of UI, steps, orchestrators, or domain concepts.
80 Args:
81 config_obj: Config dataclass instance to resolve attribute from
82 attr_name: Attribute name to resolve (e.g., 'enabled')
83 context_stack: List of context objects to resolve through (e.g., [global_config, pipeline_config, step])
84 live_context: Live values from form managers, keyed by type
85 cache_token: Current cache token for invalidation
87 Returns:
88 Resolved attribute value
89 """
90 if cache_token != self._answers_token:
91 self._answers_by_attr = {}
92 self._answers_token = cache_token
94 answers = self._answers_by_attr.setdefault(attr_name, [])
95 for answer in answers:
96 if answer.computed_from(config_obj, context_stack):
97 return answer.value
99 resolved_value = self._resolve_uncached(
100 config_obj, attr_name, context_stack, live_context
101 )
102 answers.append(_ResolvedAttr(config_obj, tuple(context_stack), resolved_value))
103 return resolved_value
105 def invalidate(self) -> None:
106 """Drop every kept answer."""
107 self._answers_token = None
108 self._answers_by_attr = {}
110 def reconstruct_live_values(self, live_values: dict[str, Any]) -> dict[str, Any]:
111 """Return live values unchanged."""
112 return live_values if live_values else {}
114 def _resolve_uncached(
115 self,
116 config_obj: object,
117 attr_name: str,
118 context_stack: list,
119 live_context: dict[type, dict[str, Any]]
120 ) -> Any:
121 """Resolve config attribute through context hierarchy (uncached)."""
122 # Merge live values into each context object
123 merged_contexts = [
124 self._merge_live_values(ctx, live_context.get(type(ctx)))
125 for ctx in context_stack
126 ]
128 # Resolve through nested context stack
129 return self._resolve_through_contexts(merged_contexts, config_obj, attr_name)
131 def _resolve_through_contexts(self, merged_contexts: list, config_obj: object, attr_name: str) -> Any:
132 """Resolve through nested context stack using config_context()."""
133 # Build nested context managers
134 if not merged_contexts:
135 # No context - just get attribute directly
136 return getattr(config_obj, attr_name)
138 # Nest contexts from outermost to innermost
139 def resolve_in_context(contexts_remaining):
140 if not contexts_remaining:
141 # Innermost level - get the attribute
142 return getattr(config_obj, attr_name)
144 # Enter context and recurse
145 ctx = contexts_remaining[0]
146 with config_context(ctx):
147 return resolve_in_context(contexts_remaining[1:])
149 return resolve_in_context(merged_contexts)
151 def _merge_live_values(self, base_obj: object, live_values: dict[str, Any] | None) -> object:
152 """Merge live values into base object.
154 CRITICAL: Passes None values through to dataclasses.replace(). When a field is reset
155 to None in a form, the None value should override the saved concrete value in the
156 base object. This allows the lazy resolution system to walk up the MRO to find the
157 inherited value from parent context.
159 Example: PipelineConfig form resets partition_filter_config.enabled to None
160 → dataclasses.replace(saved_pipeline_config,
161 partition_filter_config=LazyPartitionFilterConfig(enabled=None))
162 → When resolving enabled, the None triggers MRO walk to GlobalPipelineConfig
163 """
164 if live_values is None or not is_dataclass(base_obj):
165 return base_obj
167 # Reconstruct nested dataclasses recursively
168 reconstructed_values = self.reconstruct_live_values(live_values)
170 # Merge into base object (including None values to override saved concrete values)
171 # Use replace_raw to preserve None values (dataclasses.replace triggers lazy resolution)
172 if reconstructed_values:
173 return replace_raw(base_obj, **reconstructed_values)
174 else:
175 return base_obj