Coverage for sygaldry / resolver.py: 81%
212 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-20 01:40 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import inspect
6import warnings
7from dataclasses import dataclass
8from typing import Any, Iterable, Optional
10from .cache import Instances
11from .errors import (
12 CircularReferenceError,
13 ConfigReferenceError,
14 ConstructorError,
15 ImportResolutionError,
16 ResolutionError,
17 ValidationError,
18)
19from .types import RESERVED_KEYS, import_dotted_path
22@dataclass(frozen=True)
23class Reference:
24 """
25 Reference to a dotted config path.
26 """
28 path: str
31@dataclass(frozen=True)
32class ConstructorSpec:
33 """
34 Constructor specification for a component.
35 """
37 target: str
38 args: tuple[Any, ...]
39 kwargs: dict[str, Any]
40 instance: Optional[str]
43@dataclass(frozen=True)
44class Node:
45 """
46 Intermediate node representation for optional IR usage.
47 """
49 name: str
50 children: tuple[str, ...]
51 bindings: dict[str, Any]
54def resolve_config(
55 config: dict[str, Any],
56 *,
57 file_path: Optional[str] = None,
58 cache: Optional[Instances] = None,
59 transient: bool = False,
60) -> dict[str, Any]:
61 """
62 Resolve a config mapping into an instantiated object graph.
64 :param config: Configuration mapping.
65 :param file_path: Source config path for context.
66 :param cache: Optional instance cache.
67 :param transient: If True, bypass caching.
68 :type config: dict[str, Any]
69 :type file_path: str | None
70 :type cache: Instances | None
71 :type transient: bool
72 :returns: Resolved configuration mapping.
73 :rtype: dict[str, Any]
74 """
75 resolver = _Resolver(config, file_path=file_path, cache=cache, transient=transient)
76 return resolver.resolve()
79class _Resolver:
80 """
81 Resolver that walks and instantiates config values.
82 """
84 def __init__(
85 self,
86 config: dict[str, Any],
87 *,
88 file_path: Optional[str],
89 cache: Optional[Instances],
90 transient: bool,
91 ) -> None:
92 """
93 Initialize the resolver.
95 :param config: Configuration mapping.
96 :param file_path: Source config path for context.
97 :param cache: Optional instance cache.
98 :param transient: If True, bypass caching.
99 :type config: dict[str, Any]
100 :type file_path: str | None
101 :type cache: Instances | None
102 :type transient: bool
103 """
104 self._config = config
105 self._file_path = file_path
106 self._cache = cache or Instances()
107 self._transient = transient
108 self._resolved_top: dict[str, Any] = dict()
109 self._resolving_top: list[str] = list()
111 def resolve(self) -> dict[str, Any]:
112 """
113 Validate and resolve the full config mapping.
115 :returns: Resolved configuration mapping.
116 :rtype: dict[str, Any]
117 """
118 self._validate_schema(self._config, path=[])
119 self._validate_refs(self._config)
120 return self._resolve_value(self._config, path=[])
122 def _validate_schema(self, value: Any, *, path: list[str]) -> None:
123 """
124 Validate schema rules for reserved keys.
126 :param value: Value to validate.
127 :param path: Dotted path context.
128 :type value: object
129 :type path: list[str]
130 :raises ValidationError: On schema violations.
131 """
132 if isinstance(value, dict):
133 if "_ref" in value and len(value) != 1:
134 raise ValidationError(
135 "_ref must be the only key in its mapping.",
136 file_path=self._file_path,
137 config_path=".".join(path),
138 )
139 if "_type" in value and "_func" in value: 139 ↛ 140line 139 didn't jump to line 140 because the condition on line 139 was never true
140 raise ValidationError(
141 "_type and _func are mutually exclusive.",
142 file_path=self._file_path,
143 config_path=".".join(path),
144 )
145 if "_instance" in value and "_type" not in value: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 raise ValidationError(
147 "_instance is only valid with _type.",
148 file_path=self._file_path,
149 config_path=".".join(path),
150 )
151 for key, child in value.items():
152 next_path = path + [str(key)]
153 self._validate_schema(child, path=next_path)
154 elif isinstance(value, list):
155 for idx, child in enumerate(value):
156 self._validate_schema(child, path=path + [str(idx)])
157 elif isinstance(value, tuple): 157 ↛ 158line 157 didn't jump to line 158 because the condition on line 157 was never true
158 for idx, child in enumerate(value):
159 self._validate_schema(child, path=path + [str(idx)])
161 def _validate_refs(self, config: dict[str, Any]) -> None:
162 """
163 Validate that all _ref targets exist in the config tree.
165 For plain config paths (no ``_type`` at intermediate levels),
166 validates the full dotted path through the raw config dicts.
167 Stops validation at dict boundaries since deeper segments
168 may be attribute accesses on constructed objects.
170 :param config: Configuration mapping.
171 :type config: dict[str, Any]
172 :raises ConfigReferenceError: If any target is missing.
173 """
174 for ref_path, ref_key in _collect_refs(config):
175 top = ref_key.split(".", 1)[0]
176 if top not in config:
177 raise ConfigReferenceError(
178 f"Reference target '{ref_key}' not found.",
179 file_path=self._file_path,
180 config_path=ref_path,
181 )
182 current = config[top]
183 segments = ref_key.split(".")[1:]
184 checked = top
185 for segment in segments:
186 if not isinstance(current, dict): 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 break
188 if "_type" in current or "_func" in current:
189 break
190 if segment not in current: 190 ↛ 197line 190 didn't jump to line 197 because the condition on line 190 was always true
191 raise ConfigReferenceError(
192 f"Reference target '{ref_key}' not found "
193 f"('{segment}' missing under '{checked}').",
194 file_path=self._file_path,
195 config_path=ref_path,
196 )
197 current = current[segment]
198 checked = f"{checked}.{segment}"
200 def _resolve_value(self, value: Any, *, path: list[str]) -> Any:
201 """
202 Resolve any value recursively.
204 :param value: Value to resolve.
205 :param path: Dotted path context.
206 :type value: object
207 :type path: list[str]
208 :returns: Resolved value.
209 :rtype: object
210 """
211 if isinstance(value, dict):
212 return self._resolve_dict(value, path=path)
213 if isinstance(value, list):
214 return [
215 self._resolve_value(item, path=path + [str(idx)])
216 for idx, item in enumerate(value)
217 ]
218 if isinstance(value, tuple): 218 ↛ 219line 218 didn't jump to line 219 because the condition on line 218 was never true
219 return tuple(
220 self._resolve_value(item, path=path + [str(idx)])
221 for idx, item in enumerate(value)
222 )
223 return value
225 def _resolve_dict(self, value: dict[str, Any], *, path: list[str]) -> Any:
226 """
227 Resolve a dict, applying reserved-key rules.
229 :param value: Mapping to resolve.
230 :param path: Dotted path context.
231 :type value: dict[str, Any]
232 :type path: list[str]
233 :returns: Resolved mapping or constructed object.
234 :rtype: object
235 """
236 if "_ref" in value:
237 return self._resolve_ref(value["_ref"], path)
238 if "_type" in value:
239 return self._resolve_type(value, path)
240 if "_func" in value:
241 return self._resolve_func(value, path)
242 if value.get("_deep") is False:
243 return {k: v for k, v in value.items() if k != "_deep"}
244 if "_entries" in value:
245 return self._resolve_entries(value, path)
247 resolved: dict[str, Any] = dict()
248 for key, child in value.items():
249 resolved[key] = self._resolve_value(child, path=path + [str(key)])
250 return resolved
252 def _resolve_ref(self, ref: Any, path: list[str]) -> Any:
253 """
254 Resolve a ``_ref`` mapping.
256 :param ref: Reference value.
257 :param path: Dotted path context.
258 :type ref: object
259 :type path: list[str]
260 :returns: Resolved referenced object or attribute.
261 :rtype: object
262 :raises ConfigReferenceError: If the ref is invalid or missing.
263 """
264 if not isinstance(ref, str) or not ref: 264 ↛ 265line 264 didn't jump to line 265 because the condition on line 264 was never true
265 raise ConfigReferenceError(
266 "_ref must be a non-empty string.",
267 file_path=self._file_path,
268 config_path=".".join(path),
269 )
270 top, _, remainder = ref.partition(".")
271 target = self._resolve_top_level(top, path)
272 if remainder:
273 for attr in remainder.split("."):
274 try:
275 target = getattr(target, attr)
276 except AttributeError as exc:
277 raise ConfigReferenceError(
278 f"Attribute '{attr}' not found on reference '{top}'.",
279 file_path=self._file_path,
280 config_path=".".join(path),
281 ) from exc
282 return target
284 def _resolve_top_level(self, key: str, path: list[str]) -> Any:
285 """
286 Resolve a top-level config entry, with cycle detection.
288 :param key: Top-level key to resolve.
289 :param path: Dotted path context.
290 :type key: str
291 :type path: list[str]
292 :returns: Resolved value.
293 :rtype: object
294 :raises CircularReferenceError: If a cycle is detected.
295 """
296 if key in self._resolved_top:
297 return self._resolved_top[key]
298 if key in self._resolving_top:
299 chain = " -> ".join(self._resolving_top + [key])
300 raise CircularReferenceError(
301 f"Circular reference detected: {chain}",
302 file_path=self._file_path,
303 config_path=".".join(path),
304 )
305 if key not in self._config: 305 ↛ 306line 305 didn't jump to line 306 because the condition on line 305 was never true
306 raise ConfigReferenceError(
307 f"Reference target '{key}' not found.",
308 file_path=self._file_path,
309 config_path=".".join(path),
310 )
311 self._resolving_top.append(key)
312 resolved = self._resolve_value(self._config[key], path=[key])
313 self._resolving_top.pop()
314 self._resolved_top[key] = resolved
315 return resolved
317 def _resolve_entries(self, value: dict[str, Any], path: list[str]) -> dict[Any, Any]:
318 """
319 Resolve a mapping encoded via ``_entries``.
321 :param value: Mapping containing ``_entries`` list.
322 :param path: Dotted path context.
323 :type value: dict[str, Any]
324 :type path: list[str]
325 :returns: Resolved mapping.
326 :rtype: dict[Any, Any]
327 :raises ResolutionError: If entries are malformed.
328 """
329 entries = value.get("_entries")
330 if not isinstance(entries, list): 330 ↛ 331line 330 didn't jump to line 331 because the condition on line 330 was never true
331 raise ResolutionError(
332 "_entries must be a list.",
333 file_path=self._file_path,
334 config_path=".".join(path),
335 )
336 if len(value.keys() - {"_entries"}) != 0: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 raise ResolutionError(
338 "_entries cannot be combined with other keys.",
339 file_path=self._file_path,
340 config_path=".".join(path),
341 )
342 resolved: dict[Any, Any] = dict()
343 for idx, entry in enumerate(entries):
344 if not isinstance(entry, dict) or "_key" not in entry or "_value" not in entry: 344 ↛ 345line 344 didn't jump to line 345 because the condition on line 344 was never true
345 raise ResolutionError(
346 "Each _entries item must contain _key and _value.",
347 file_path=self._file_path,
348 config_path=".".join(path + [str(idx)]),
349 )
350 key = self._resolve_value(entry["_key"], path=path + [str(idx), "_key"])
351 val = self._resolve_value(entry["_value"], path=path + [str(idx), "_value"])
352 try:
353 hash(key)
354 except Exception as exc:
355 raise ResolutionError(
356 "Resolved _entries keys must be hashable.",
357 file_path=self._file_path,
358 config_path=".".join(path + [str(idx), "_key"]),
359 ) from exc
360 if key in resolved: 360 ↛ 361line 360 didn't jump to line 361 because the condition on line 360 was never true
361 raise ResolutionError(
362 f"Duplicate _entries key '{key}'.",
363 file_path=self._file_path,
364 config_path=".".join(path + [str(idx), "_key"]),
365 )
366 resolved[key] = val
367 return resolved
369 def _resolve_func(self, value: dict[str, Any], path: list[str]) -> Any:
370 """
371 Resolve a ``_func`` mapping to a callable.
373 :param value: Mapping containing ``_func``.
374 :param path: Dotted path context.
375 :type value: dict[str, Any]
376 :type path: list[str]
377 :returns: Imported callable.
378 :rtype: object
379 :raises ValidationError: If extra keys accompany ``_func``.
380 """
381 extras = set(value) - {"_func"}
382 if extras:
383 raise ValidationError(
384 f"_func does not accept other keys, found: {sorted(extras)}.",
385 file_path=self._file_path,
386 config_path=".".join(path),
387 )
388 func_path = value.get("_func")
389 try:
390 return import_dotted_path(
391 func_path,
392 file_path=self._file_path,
393 config_path=".".join(path),
394 )
395 except ImportResolutionError:
396 raise
398 def _resolve_type(self, value: dict[str, Any], path: list[str]) -> Any:
399 """
400 Resolve a ``_type`` mapping into an instance.
402 :param value: Mapping containing ``_type`` and constructor data.
403 :param path: Dotted path context.
404 :type value: dict[str, Any]
405 :type path: list[str]
406 :returns: Constructed instance (possibly cached).
407 :rtype: object
408 """
409 type_path = value.get("_type")
410 args = value.get("_args", [])
411 instance = value.get("_instance")
412 if not isinstance(args, list): 412 ↛ 413line 412 didn't jump to line 413 because the condition on line 412 was never true
413 raise ResolutionError(
414 "_args must be a list.",
415 file_path=self._file_path,
416 config_path=".".join(path),
417 )
419 kwargs = {k: v for k, v in value.items() if k not in RESERVED_KEYS}
420 resolved_args = tuple(self._resolve_value(arg, path=path + ["_args"]) for arg in args)
421 resolved_kwargs = {
422 key: self._resolve_value(val, path=path + [key]) for key, val in kwargs.items()
423 }
425 try:
426 target = import_dotted_path(
427 type_path,
428 file_path=self._file_path,
429 config_path=".".join(path),
430 )
431 except ImportResolutionError:
432 raise
434 resolved_kwargs = _validate_signature(
435 target,
436 resolved_args,
437 resolved_kwargs,
438 file_path=self._file_path,
439 config_path=".".join(path),
440 )
442 def factory() -> Any:
443 try:
444 return target(*resolved_args, **resolved_kwargs)
445 except Exception as exc: # noqa: BLE001
446 raise ConstructorError(
447 f"Failed to construct '{type_path}'.",
448 file_path=self._file_path,
449 config_path=".".join(path),
450 ) from exc
452 return self._cache.get_or_create(
453 type_path,
454 instance,
455 resolved_args,
456 resolved_kwargs,
457 factory,
458 transient=self._transient,
459 file_path=self._file_path,
460 config_path=".".join(path),
461 )
464def _collect_refs(
465 value: Any, *, path: Optional[list[str]] = None
466) -> Iterable[tuple[str, str]]:
467 """
468 Collect ``_ref`` occurrences from a config tree.
470 :param value: Root value to inspect.
471 :param path: Current path stack.
472 :type value: object
473 :type path: list[str] | None
474 :returns: Iterable of (config_path, ref_target) pairs.
475 :rtype: collections.abc.Iterable[tuple[str, str]]
476 """
477 path = path or list()
478 if isinstance(value, dict):
479 if "_ref" in value:
480 yield ".".join(path), value["_ref"]
481 return
482 for key, child in value.items():
483 yield from _collect_refs(child, path=path + [str(key)])
484 elif isinstance(value, list):
485 for idx, child in enumerate(value):
486 yield from _collect_refs(child, path=path + [str(idx)])
487 elif isinstance(value, tuple): 487 ↛ 488line 487 didn't jump to line 488 because the condition on line 487 was never true
488 for idx, child in enumerate(value):
489 yield from _collect_refs(child, path=path + [str(idx)])
492def _validate_signature(
493 target: Any,
494 args: tuple[Any, ...],
495 kwargs: dict[str, Any],
496 *,
497 file_path: Optional[str],
498 config_path: str,
499) -> dict[str, Any]:
500 """
501 Validate constructor signature and filter extra kwargs.
503 :param target: Callable to validate.
504 :param args: Positional arguments.
505 :param kwargs: Keyword arguments.
506 :param file_path: Source config path for context.
507 :param config_path: Dotted config path for context.
508 :type target: object
509 :type args: tuple[Any, ...]
510 :type kwargs: dict[str, Any]
511 :type file_path: str | None
512 :type config_path: str
513 :returns: Possibly filtered kwargs.
514 :rtype: dict[str, Any]
515 :raises ConstructorError: If required parameters are missing.
516 """
517 try:
518 signature = inspect.signature(target)
519 except (TypeError, ValueError):
520 return kwargs
522 has_var_kw = any(p.kind == p.VAR_KEYWORD for p in signature.parameters.values())
523 if not has_var_kw: 523 ↛ 538line 523 didn't jump to line 538 because the condition on line 523 was always true
524 allowed = {
525 name
526 for name, param in signature.parameters.items()
527 if param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
528 }
529 extras = set(kwargs) - allowed
530 if extras: 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true
531 warnings.warn(
532 f"Extra kwargs for '{getattr(target, '__name__', target)}': {sorted(extras)}",
533 RuntimeWarning,
534 stacklevel=2,
535 )
536 kwargs = {k: v for k, v in kwargs.items() if k not in extras}
538 try:
539 bound = signature.bind_partial(*args, **kwargs)
540 except TypeError as exc:
541 raise ConstructorError(
542 f"Constructor signature mismatch for '{getattr(target, '__name__', target)}': {exc}.",
543 file_path=file_path,
544 config_path=config_path,
545 ) from exc
547 missing = list()
548 for name, param in signature.parameters.items():
549 if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD): 549 ↛ 550line 549 didn't jump to line 550 because the condition on line 549 was never true
550 continue
551 if param.default is not param.empty:
552 continue
553 if name not in bound.arguments: 553 ↛ 554line 553 didn't jump to line 554 because the condition on line 553 was never true
554 missing.append(name)
556 if missing: 556 ↛ 557line 556 didn't jump to line 557 because the condition on line 556 was never true
557 raise ConstructorError(
558 f"Missing required parameters: {missing}.",
559 file_path=file_path,
560 config_path=config_path,
561 )
562 return kwargs
565if __name__ == "__main__": 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 pass