Coverage for src/objectstate/dual_axis_resolver.py: 94%
98 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-17 07:22 +0000
1"""
2Generic dual-axis resolver for lazy configuration inheritance.
4``resolve_with_provenance`` is the one resolution of a field in the active context,
5over both the context hierarchy (X-axis) and sibling inheritance (Y-axis). It returns
6the value together with the scope and type that supplied it.
8The resolver is completely generic and has no application-specific dependencies.
9"""
11import functools
12import logging
13from dataclasses import MISSING, fields, is_dataclass
14from typing import Any, NamedTuple
16logger = logging.getLogger(__name__)
19class ResolvedField(NamedTuple):
20 """What a field resolves to in one context, with the layer that supplied it.
22 ``source_scope`` is the scope id of the supplying layer and ``source_type`` the
23 config type it was read from, which is the container type or one of its bases.
25 When no layer holds a value, ``value`` is the declared default (or ``None``) and
26 the source names where the field lives in the context: the outermost layer holding
27 the nearest type in the MRO. Both are ``None`` when no layer holds such a type.
28 """
30 value: Any
31 source_scope: str | None
32 source_type: type | None
35@functools.lru_cache(maxsize=None)
36def _normalize_to_base(t: type) -> type:
37 """Normalize lazy type to its base type for comparison.
39 LazyPartitionFilterConfig -> PartitionFilterConfig
40 PartitionFilterConfig -> PartitionFilterConfig
42 PERFORMANCE: Results are cached with lru_cache since type->base_type
43 mappings are immutable after class creation. Import is hoisted to
44 avoid per-call import overhead.
45 """
46 from objectstate.lazy_factory import get_base_type_for_lazy
47 return get_base_type_for_lazy(t) or t
50def _direct_owner_field_default(config_type: type, field_name: str) -> tuple[bool, Any]:
51 """Return the default owned by config_type for field_name.
53 Lazy config resolution intentionally forces inherited dataclass fields to
54 None so they can inherit through context and MRO. Fields declared directly
55 on the concrete owner still have their semantic class defaults on the base
56 config type. Those defaults are load-bearing when no scope supplies a
57 concrete value.
58 """
59 if not is_dataclass(config_type):
60 return False, None
62 field_obj = config_type.__dataclass_fields__.get(field_name)
63 if field_obj is None:
64 return False, None
66 metadata = field_obj.metadata or {}
67 if "_inherited_default" in metadata or "_inherited_default_factory" in metadata:
68 return False, None
70 if field_obj.default is not MISSING:
71 return True, field_obj.default
73 if field_obj.default_factory is not MISSING:
74 return True, field_obj.default_factory()
76 return False, None
79def declared_default(container_type: type, field_name: str) -> tuple[Any, type | None]:
80 """The value a field has when no context layer supplies one.
82 In order: the default the base type declares for the field itself, the first
83 concrete class attribute along the base type's MRO, then the default the field
84 had before it was rebuilt to inherit (kept in the field's metadata).
86 Returns the value and, when the base type declares a default for the field
87 itself, that base type; otherwise ``None`` in its place.
88 """
89 container_base = _normalize_to_base(container_type)
90 has_own_default, own_default = _direct_owner_field_default(container_base, field_name)
91 own_default_type = container_base if has_own_default else None
92 if own_default is not None:
93 return own_default, own_default_type
95 for mro_class in container_base.__mro__:
96 class_value = getattr(mro_class, field_name, None)
97 if class_value is not None:
98 return class_value, own_default_type
100 if is_dataclass(container_type):
101 for field_definition in fields(container_type):
102 if field_definition.name != field_name:
103 continue
104 metadata = field_definition.metadata
105 inherited = metadata.get("_inherited_default", MISSING)
106 if inherited is not MISSING:
107 return inherited, own_default_type
108 factory = metadata.get("_inherited_default_factory", MISSING)
109 if factory is not MISSING:
110 return factory(), own_default_type
111 return None, own_default_type
114def resolve_with_provenance(container_type: type, field_name: str) -> ResolvedField:
115 """
116 Resolve a field in the active context: its value and the layer that supplied it.
118 This is the one resolution. Attribute access on a lazy config takes ``.value``
119 from it; ``ObjectState`` takes the value and the source together.
121 The answer is kept on the active layer stack, so it lives exactly as long as the
122 context it was computed from.
124 Args:
125 container_type: The type containing the field (e.g., LazyPathPlanningConfig)
126 field_name: Name of the field to resolve (e.g., "partition_filter")
128 Raises:
129 NoActiveContextError: no ``config_context`` is active.
130 """
131 from objectstate.context_manager import NoActiveContextError, context_layer_stack
133 stack = context_layer_stack.get()
134 if not stack.layers:
135 raise NoActiveContextError(
136 f"Cannot resolve {container_type.__name__}.{field_name}: "
137 "no config context is active."
138 )
139 return resolve_in_stack(stack, container_type, field_name)
142def resolve_in_stack(stack, container_type: type, field_name: str) -> ResolvedField:
143 """``resolve_with_provenance`` for a layer stack the caller already holds."""
144 key = (container_type, field_name)
145 resolved = stack.resolved_fields.get(key)
146 if resolved is None:
147 resolved = stack.resolved_fields[key] = _resolve_in_layers(
148 stack.layers, container_type, field_name
149 )
150 return resolved
153def _resolve_in_layers(layers: tuple, container_type: type, field_name: str) -> ResolvedField:
154 """Resolve class first: the MRO is the outer loop, the layers the inner one.
156 For each class in the container type's MRO, most specific first, every layer is
157 asked from the innermost scope to the outermost; the first value that is not
158 ``None`` wins. ``None`` in an inner layer means "inherit", so a less specific
159 class is tried only after every layer was asked for the more specific one.
160 A layer that masks the layers outside it ends the walk outward.
162 The order is stated with an example in ``docs/context_system.rst`` under
163 "Precedence Order"; this function is its implementation.
164 """
165 container_base = _normalize_to_base(container_type)
167 visible_layers = []
168 for layer in reversed(layers):
169 visible_layers.append(layer)
170 if layer.masks_outer_layers:
171 break
173 # Where the field lives when every layer leaves it None: the outermost layer
174 # holding the first MRO type that any layer holds.
175 unset_scope: str | None = None
176 unset_type: type | None = None
177 unset_found = False
179 for mro_class in container_base.__mro__:
180 if not is_dataclass(mro_class):
181 continue
182 mro_type = _normalize_to_base(mro_class)
183 outermost_scope: str | None = None
184 held_by_a_layer = False
186 for layer in visible_layers:
187 config_instance = layer.configs.get(mro_type)
188 if config_instance is None:
189 continue
190 held_by_a_layer = True
191 outermost_scope = layer.scope_id
192 try:
193 value = object.__getattribute__(config_instance, field_name)
194 except AttributeError:
195 continue
196 if value is not None:
197 return ResolvedField(value, layer.scope_id, mro_type)
199 if held_by_a_layer and not unset_found:
200 unset_found = True
201 unset_scope = outermost_scope
202 unset_type = mro_type
204 default_value, own_default_type = declared_default(container_type, field_name)
205 return ResolvedField(default_value, unset_scope, own_default_type or unset_type)
208def get_field_provenance(
209 container_type: type, field_name: str
210) -> tuple[str | None, type | None]:
211 """
212 Find which scope AND type provided the concrete value for a field.
214 The scope and type of ``resolve_with_provenance()``, without the value.
216 Args:
217 container_type: The type containing the field (e.g., LazyPathPlanningConfig)
218 field_name: Name of the field to find provenance for (e.g., "partition_filter")
220 Returns:
221 (source_scope_id, source_type): The scope_id and type that provided the value.
222 source_type may differ from container_type due to MRO inheritance.
223 See ``ResolvedField`` for what they name when no layer has a concrete value.
224 """
225 _, source_scope, source_type = resolve_with_provenance(container_type, field_name)
226 return source_scope, source_type