Coverage for sygaldry / artificery.py: 88%
279 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-06 19:06 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import inspect
6import warnings
7from collections.abc import Iterable
8from dataclasses import dataclass
9from pathlib import Path
10from typing import Any, Optional
12from .cache import Instances
13from .errors import (
14 CircularReferenceError,
15 ConfigReferenceError,
16 ConstructorError,
17 ImportResolutionError,
18 ResolutionError,
19 ValidationError,
20)
21from .loader import (
22 _MISSING,
23 _deep_merge,
24 _get_by_path,
25 _interpolate_config,
26 _load_with_includes,
27 load_config,
28)
29from .types import RESERVED_KEYS, import_dotted_path
32@dataclass(frozen=True)
33class Reference:
34 """
35 Reference to a dotted config path.
36 """
38 path: str
41@dataclass(frozen=True)
42class ConstructorSpec:
43 """
44 Constructor specification for a component.
45 """
47 target: str
48 args: tuple[Any, ...]
49 kwargs: dict[str, Any]
50 instance: str | None
53@dataclass(frozen=True)
54class Node:
55 """
56 Intermediate node representation for optional IR usage.
57 """
59 name: str
60 children: tuple[str, ...]
61 bindings: dict[str, Any]
64def _set_by_path(data: dict[str, Any], dotted_path: str, value: Any) -> None:
65 """
66 Set a nested value by dotted path, creating intermediate dicts.
68 :param data: Root mapping to modify in-place.
69 :param dotted_path: Dotted key path (e.g. ``"db.host"``).
70 :param value: Value to set.
71 :type data: dict
72 :type dotted_path: str
73 :type value: object
74 """
75 segments = dotted_path.split(".")
76 current = data
77 for segment in segments[:-1]:
78 if segment not in current or not isinstance(current[segment], dict):
79 current[segment] = dict()
80 current = current[segment]
81 current[segments[-1]] = value
84def load_config_file(path: str | Path) -> dict[str, Any]:
85 """
86 Load a config file without resolving components.
88 :param path: Path to a YAML or TOML config file.
89 :type path: str | pathlib.Path
90 :returns: Parsed and interpolated mapping.
91 :rtype: dict[str, Any]
92 """
93 file_path = Path(path)
94 return load_config(file_path)
97class ArtificeryLoader:
98 """
99 Lightweight loader wrapper for config files.
100 """
102 def __init__(self, path: str | Path) -> None:
103 """
104 Initialize the loader.
106 :param path: Path to a YAML or TOML config file.
107 :type path: str | pathlib.Path
108 """
109 self._path = Path(path)
111 def load(self) -> dict[str, Any]:
112 """
113 Load and return the config mapping.
115 :returns: Parsed and interpolated configuration mapping.
116 :rtype: dict[str, Any]
117 """
118 return load_config(self._path)
121def resolve_config(
122 config: dict[str, Any],
123 *,
124 file_path: str | None = None,
125 cache: Instances | None = None,
126 transient: bool = False,
127) -> dict[str, Any]:
128 """
129 Resolve a config mapping into an instantiated object graph.
131 :param config: Configuration mapping.
132 :param file_path: Source config path for context.
133 :param cache: Optional instance cache.
134 :param transient: If True, bypass caching.
135 :type config: dict[str, Any]
136 :type file_path: str | None
137 :type cache: Instances | None
138 :type transient: bool
139 :returns: Resolved configuration mapping.
140 :rtype: dict[str, Any]
141 """
143 artificery = Artificery(config=config, source=file_path, cache=cache, transient=transient)
144 resolved = artificery.resolve()
145 return resolved
148class Artificery:
149 """
150 Component factory that loads, merges, and resolves config into objects.
151 """
153 def __init__(
154 self,
155 *paths: str | Path,
156 config: dict[str, Any] | None = None,
157 source: str | None = None,
158 overrides: dict[str, Any] | None = None,
159 uses: dict[str, str] | None = None,
160 cache: Instances | None = None,
161 transient: bool = False,
162 ) -> None:
163 """
164 Initialize the Artificery.
166 :param paths: Config file paths to load and deep-merge in order.
167 :param config: Pre-loaded configuration mapping.
168 :param source: Label for error messages (auto-derived from paths if omitted).
169 :param overrides: Dotted-path overrides applied after merging, before interpolation.
170 :param uses: Dotted-path mappings (target -> source) copied within the merged config.
171 :param cache: Optional instance cache.
172 :param transient: If True, bypass caching.
173 :type paths: str | pathlib.Path
174 :type config: dict | None
175 :type source: str | None
176 :type overrides: dict[str, Any] | None
177 :type uses: dict[str, str] | None
178 :type cache: Instances | None
179 :type transient: bool
180 :raises ValueError: If no config source is provided.
181 """
182 if not paths and config is None:
183 raise ValueError("Artificery requires at least one path or a config dict.")
185 self._active_config: dict[str, Any] = dict()
186 self._cache = cache or Instances()
187 self._overrides = overrides or dict()
188 self._paths = tuple(Path(path) for path in paths)
189 self._prepared: dict[str, Any] | None = None
190 self._raw_config = config
191 self._resolved_top: dict[str, Any] = dict()
192 self._resolving_top: list[str] = list()
193 self._source = source
194 self._uses = uses or dict()
195 self._transient = transient
197 @property
198 def config(self) -> dict[str, Any]:
199 """
200 The loaded, merged, and interpolated configuration (before resolution).
202 :returns: Prepared configuration mapping.
203 :rtype: dict[str, Any]
204 """
205 if not self._prepared:
206 self._prepared = self._prepare()
207 return self._prepared
209 @property
210 def _source_label(self) -> str | None:
211 """
212 Derive a source label for error messages.
213 """
214 if self._source is not None: 214 ↛ 215line 214 didn't jump to line 215 because the condition on line 214 was never true
215 return self._source
216 elif self._paths:
217 return str(self._paths[-1])
218 else:
219 return None
221 def _prepare(self) -> dict[str, Any]:
222 """
223 Load, merge, apply overrides, and interpolate the config.
225 :returns: Prepared configuration mapping.
226 :rtype: dict[str, Any]
227 """
228 if self._paths:
229 merged = self._load_and_merge()
230 if self._raw_config is not None: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 merged = _deep_merge(merged, self._raw_config)
232 merged = self._apply_uses(merged)
233 merged = self._apply_overrides(merged)
234 return _interpolate_config(merged, file_path=self._source_label)
235 elif self._raw_config is not None: 235 ↛ 238line 235 didn't jump to line 238 because the condition on line 235 was always true
236 return self._raw_config
237 else:
238 raise ValueError("Artificery has no config source.")
240 def _load_and_merge(self) -> dict[str, Any]:
241 """
242 Load multiple config files and deep-merge in order.
244 :returns: Merged raw config mapping.
245 :rtype: dict[str, Any]
246 """
247 merged: dict[str, Any] = dict()
248 for path in self._paths:
249 resolved_path = path.expanduser().resolve()
250 visited: set[Path] = set()
251 data = _load_with_includes(resolved_path, ancestors=[], visited=visited)
252 merged = _deep_merge(merged, data)
253 return merged
255 def _apply_uses(self, config: dict[str, Any]) -> dict[str, Any]:
256 """
257 Copy values within the config for each use mapping.
259 :param config: Merged raw config.
260 :type config: dict[str, Any]
261 :returns: Modified config.
262 :rtype: dict[str, Any]
263 """
264 for target_path, source_path in self._uses.items():
265 value = _get_by_path(config, source_path)
266 if value is _MISSING:
267 available = sorted(config.keys())
268 raise ConfigReferenceError(
269 f"Use source '{source_path}' not found. "
270 f"Available top-level keys: {available}",
271 file_path=self._source_label,
272 config_path=target_path,
273 )
274 _set_by_path(config, target_path, value)
275 return config
277 def _apply_overrides(self, config: dict[str, Any]) -> dict[str, Any]:
278 """
279 Apply dotted-path overrides to the config.
281 :param config: Merged raw config.
282 :type config: dict[str, Any]
283 :returns: Modified config.
284 :rtype: dict[str, Any]
285 """
286 for key, value in self._overrides.items():
287 _set_by_path(config, key, value)
288 return config
290 def resolve(self) -> dict[str, Any]:
291 """
292 Validate and resolve the full config mapping into objects.
294 :returns: Resolved configuration mapping.
295 :rtype: dict[str, Any]
296 """
297 active = self.config
298 self._active_config = active
299 self._resolved_top: dict[str, Any] = dict()
300 self._resolving_top: list[str] = list()
301 self._validate_schema(active, path=[])
302 self._validate_refs(active)
303 return self._resolve_value(active, path=[])
305 def _validate_schema(self, value: Any, *, path: list[str]) -> None:
306 """
307 Validate schema rules for reserved keys.
309 :param value: Value to validate.
310 :param path: Dotted path context.
311 :type value: object
312 :type path: list[str]
313 :raises ValidationError: On schema violations.
314 """
315 if isinstance(value, dict):
316 if "_ref" in value and len(value) != 1:
317 raise ValidationError(
318 "_ref must be the only key in its mapping.",
319 file_path=self._source_label,
320 config_path=".".join(path),
321 )
322 elif "_type" in value and "_func" in value: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true
323 raise ValidationError(
324 "_type and _func are mutually exclusive.",
325 file_path=self._source_label,
326 config_path=".".join(path),
327 )
328 elif "_instance" in value and "_type" not in value: 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true
329 raise ValidationError(
330 "_instance is only valid with _type.",
331 file_path=self._source_label,
332 config_path=".".join(path),
333 )
335 for key, child in value.items():
336 next_path = path + [str(key)]
337 self._validate_schema(child, path=next_path)
339 elif isinstance(value, list) or isinstance(value, tuple):
340 for idx, child in enumerate(value):
341 self._validate_schema(child, path=path + [str(idx)])
343 def _validate_refs(self, config: dict[str, Any]) -> None:
344 """
345 Validate that all _ref targets exist in the config tree.
347 For plain config paths (no ``_type`` at intermediate levels),
348 validates the full dotted path through the raw config dicts.
349 Stops validation at dict boundaries since deeper segments
350 may be attribute accesses on constructed objects.
352 :param config: Configuration mapping.
353 :type config: dict[str, Any]
354 :raises ConfigReferenceError: If any target is missing.
355 """
356 for ref_path, ref_key in _collect_refs(config):
357 top = ref_key.split(".", 1)[0]
358 if top not in config:
359 raise ConfigReferenceError(
360 f"Reference target '{ref_key}' not found.",
361 file_path=self._source_label,
362 config_path=ref_path,
363 )
364 current = config[top]
365 segments = ref_key.split(".")[1:]
366 checked = top
367 for segment in segments:
368 if not isinstance(current, dict) or "_type" in current or "_func" in current:
369 break
370 elif segment not in current: 370 ↛ 378line 370 didn't jump to line 378 because the condition on line 370 was always true
371 raise ConfigReferenceError(
372 f"Reference target '{ref_key}' not found "
373 f"('{segment}' missing under '{checked}').",
374 file_path=self._source_label,
375 config_path=ref_path,
376 )
377 else:
378 current = current[segment]
379 checked = f"{checked}.{segment}"
381 def _resolve_value(self, value: Any, *, path: list[str]) -> Any:
382 """
383 Resolve any value recursively.
385 :param value: Value to resolve.
386 :param path: Dotted path context.
387 :type value: object
388 :type path: list[str]
389 :returns: Resolved value.
390 :rtype: object
391 """
392 if isinstance(value, dict):
393 return self._resolve_dict(value, path=path)
394 elif isinstance(value, list):
395 return [
396 self._resolve_value(item, path=path + [str(idx)])
397 for idx, item in enumerate(value)
398 ]
399 elif isinstance(value, tuple): 399 ↛ 400line 399 didn't jump to line 400 because the condition on line 399 was never true
400 return tuple(
401 self._resolve_value(item, path=path + [str(idx)])
402 for idx, item in enumerate(value)
403 )
404 else:
405 return value
407 def _resolve_dict(self, value: dict[str, Any], *, path: list[str]) -> Any:
408 """
409 Resolve a dict, applying reserved-key rules.
411 :param value: Mapping to resolve.
412 :param path: Dotted path context.
413 :type value: dict[str, Any]
414 :type path: list[str]
415 :returns: Resolved mapping or constructed object.
416 :rtype: object
417 """
418 if "_ref" in value:
419 return self._resolve_ref(value["_ref"], path)
420 elif "_type" in value:
421 return self._resolve_type(value, path)
422 elif "_func" in value:
423 return self._resolve_func(value, path)
424 elif value.get("_deep") is False:
425 return {k: v for k, v in value.items() if k != "_deep"}
426 elif "_entries" in value:
427 return self._resolve_entries(value, path)
428 else:
429 resolved: dict[str, Any] = dict()
430 for key, child in value.items():
431 resolved[key] = self._resolve_value(child, path=path + [str(key)])
432 return resolved
434 def _resolve_ref(self, ref: Any, path: list[str]) -> Any:
435 """
436 Resolve a ``_ref`` mapping.
438 :param ref: Reference value.
439 :param path: Dotted path context.
440 :type ref: object
441 :type path: list[str]
442 :returns: Resolved referenced object or attribute.
443 :rtype: object
444 :raises ConfigReferenceError: If the ref is invalid or missing.
445 """
446 if not isinstance(ref, str) or not ref: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true
447 raise ConfigReferenceError(
448 "_ref must be a non-empty string.",
449 file_path=self._source_label,
450 config_path=".".join(path),
451 )
452 else:
453 top, _, remainder = ref.partition(".")
454 target = self._resolve_top_level(top, path)
455 if remainder:
456 for attr in remainder.split("."):
457 try:
458 target = getattr(target, attr)
459 except AttributeError as exc:
460 raise ConfigReferenceError(
461 f"Attribute '{attr}' not found on reference '{top}'.",
462 file_path=self._source_label,
463 config_path=".".join(path),
464 ) from exc
465 return target
467 def _resolve_top_level(self, key: str, path: list[str]) -> Any:
468 """
469 Resolve a top-level config entry, with cycle detection.
471 :param key: Top-level key to resolve.
472 :param path: Dotted path context.
473 :type key: str
474 :type path: list[str]
475 :returns: Resolved value.
476 :rtype: object
477 :raises CircularReferenceError: If a cycle is detected.
478 """
479 if key in self._resolved_top:
480 return self._resolved_top[key]
481 elif key in self._resolving_top:
482 chain = " -> ".join(self._resolving_top + [key])
483 raise CircularReferenceError(
484 f"Circular reference detected: {chain}",
485 file_path=self._source_label,
486 config_path=".".join(path),
487 )
488 elif key not in self._active_config: 488 ↛ 489line 488 didn't jump to line 489 because the condition on line 488 was never true
489 raise ConfigReferenceError(
490 f"Reference target '{key}' not found.",
491 file_path=self._source_label,
492 config_path=".".join(path),
493 )
494 else:
495 self._resolving_top.append(key)
496 resolved = self._resolve_value(self._active_config[key], path=[key])
497 self._resolving_top.pop()
498 self._resolved_top[key] = resolved
499 return resolved
501 def _resolve_entries(self, value: dict[str, Any], path: list[str]) -> dict[Any, Any]:
502 """
503 Resolve a mapping encoded via ``_entries``.
505 :param value: Mapping containing ``_entries`` list.
506 :param path: Dotted path context.
507 :type value: dict[str, Any]
508 :type path: list[str]
509 :returns: Resolved mapping.
510 :rtype: dict[Any, Any]
511 :raises ResolutionError: If entries are malformed.
512 """
513 entries = value.get("_entries")
514 if not isinstance(entries, list): 514 ↛ 515line 514 didn't jump to line 515 because the condition on line 514 was never true
515 raise ResolutionError(
516 "_entries must be a list.",
517 file_path=self._source_label,
518 config_path=".".join(path),
519 )
520 elif len(value.keys() - {"_entries"}) != 0: 520 ↛ 521line 520 didn't jump to line 521 because the condition on line 520 was never true
521 raise ResolutionError(
522 "_entries cannot be combined with other keys.",
523 file_path=self._source_label,
524 config_path=".".join(path),
525 )
526 else:
527 resolved: dict[Any, Any] = dict()
528 for idx, entry in enumerate(entries):
529 if not isinstance(entry, dict) or "_key" not in entry or "_value" not in entry: 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true
530 raise ResolutionError(
531 "Each _entries item must contain _key and _value.",
532 file_path=self._source_label,
533 config_path=".".join(path + [str(idx)]),
534 )
535 key = self._resolve_value(entry["_key"], path=path + [str(idx), "_key"])
536 val = self._resolve_value(entry["_value"], path=path + [str(idx), "_value"])
537 try:
538 hash(key)
539 except Exception as exc:
540 raise ResolutionError(
541 "Resolved _entries keys must be hashable.",
542 file_path=self._source_label,
543 config_path=".".join(path + [str(idx), "_key"]),
544 ) from exc
546 if key in resolved: 546 ↛ 547line 546 didn't jump to line 547 because the condition on line 546 was never true
547 raise ResolutionError(
548 f"Duplicate _entries key '{key}'.",
549 file_path=self._source_label,
550 config_path=".".join(path + [str(idx), "_key"]),
551 )
552 else:
553 resolved[key] = val
554 return resolved
556 def _resolve_func(self, value: dict[str, Any], path: list[str]) -> Any:
557 """
558 Resolve a ``_func`` mapping to a callable.
560 :param value: Mapping containing ``_func``.
561 :param path: Dotted path context.
562 :type value: dict[str, Any]
563 :type path: list[str]
564 :returns: Imported callable.
565 :rtype: object
566 :raises ValidationError: If extra keys accompany ``_func``.
567 """
568 extras = set(value) - {"_func"}
569 if extras:
570 raise ValidationError(
571 f"_func does not accept other keys, found: {sorted(extras)}.",
572 file_path=self._source_label,
573 config_path=".".join(path),
574 )
575 else:
576 func_path = value.get("_func")
577 try:
578 return import_dotted_path(
579 func_path,
580 file_path=self._source_label,
581 config_path=".".join(path),
582 )
583 except ImportResolutionError:
584 raise
586 def _resolve_type(self, value: dict[str, Any], path: list[str]) -> Any:
587 """
588 Resolve a ``_type`` mapping into an instance.
590 :param value: Mapping containing ``_type`` and constructor data.
591 :param path: Dotted path context.
592 :type value: dict[str, Any]
593 :type path: list[str]
594 :returns: Constructed instance (possibly cached).
595 :rtype: object
596 """
597 type_path = value.get("_type")
598 args = value.get("_args", [])
599 extra_kwargs = value.get("_kwargs", {})
600 instance = value.get("_instance")
601 if not isinstance(args, list): 601 ↛ 602line 601 didn't jump to line 602 because the condition on line 601 was never true
602 raise ResolutionError(
603 "_args must be a list.",
604 file_path=self._source_label,
605 config_path=".".join(path),
606 )
607 elif not isinstance(extra_kwargs, dict):
608 raise ResolutionError(
609 "_kwargs must be a mapping.",
610 file_path=self._source_label,
611 config_path=".".join(path),
612 )
613 else:
614 kwargs = {k: v for k, v in value.items() if k not in RESERVED_KEYS}
615 resolved_args = tuple(
616 self._resolve_value(arg, path=path + ["_args"]) for arg in args
617 )
618 resolved_kwargs = {
619 key: self._resolve_value(val, path=path + [key]) for key, val in kwargs.items()
620 }
621 for key, val in extra_kwargs.items():
622 resolved_kwargs[key] = self._resolve_value(val, path=path + ["_kwargs", key])
624 target = import_dotted_path(
625 type_path,
626 file_path=self._source_label,
627 config_path=".".join(path),
628 )
630 resolved_kwargs = _validate_signature(
631 target,
632 resolved_args,
633 resolved_kwargs,
634 file_path=self._source_label,
635 config_path=".".join(path),
636 )
638 def factory() -> Any:
639 try:
640 return target(*resolved_args, **resolved_kwargs)
641 except Exception as exc: # noqa: BLE001
642 raise ConstructorError(
643 f"Failed to construct '{type_path}'.",
644 file_path=self._source_label,
645 config_path=".".join(path),
646 ) from exc
648 instance = self._cache.get_or_create(
649 type_path,
650 instance,
651 resolved_args,
652 resolved_kwargs,
653 factory,
654 transient=self._transient,
655 file_path=self._source_label,
656 config_path=".".join(path),
657 )
658 return instance
661def _collect_refs(value: Any, *, path: list[str] | None = None) -> Iterable[tuple[str, str]]:
662 """
663 Collect ``_ref`` occurrences from a config tree.
665 :param value: Root value to inspect.
666 :param path: Current path stack.
667 :type value: object
668 :type path: list[str] | None
669 :returns: Iterable of (config_path, ref_target) pairs.
670 :rtype: collections.abc.Iterable[tuple[str, str]]
671 """
672 path = path or list()
673 if isinstance(value, dict):
674 if "_ref" in value:
675 yield ".".join(path), value["_ref"]
676 return
677 else:
678 for key, child in value.items():
679 yield from _collect_refs(child, path=path + [str(key)])
680 elif isinstance(value, list) or isinstance(value, tuple):
681 for idx, child in enumerate(value):
682 yield from _collect_refs(child, path=path + [str(idx)])
685def _validate_signature(
686 target: Any,
687 args: tuple[Any, ...],
688 kwargs: dict[str, Any],
689 *,
690 file_path: str | None,
691 config_path: str,
692) -> dict[str, Any]:
693 """
694 Validate constructor signature and filter extra kwargs.
696 :param target: Callable to validate.
697 :param args: Positional arguments.
698 :param kwargs: Keyword arguments.
699 :param file_path: Source config path for context.
700 :param config_path: Dotted config path for context.
701 :type target: object
702 :type args: tuple[Any, ...]
703 :type kwargs: dict[str, Any]
704 :type file_path: str | None
705 :type config_path: str
706 :returns: Possibly filtered kwargs.
707 :rtype: dict[str, Any]
708 :raises ConstructorError: If required parameters are missing.
709 """
710 try:
711 signature = inspect.signature(target)
712 except (TypeError, ValueError):
713 return kwargs
715 has_var_kw = any(p.kind == p.VAR_KEYWORD for p in signature.parameters.values())
716 if not has_var_kw:
717 allowed = {
718 name
719 for name, param in signature.parameters.items()
720 if param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
721 }
722 extras = set(kwargs) - allowed
723 if extras: 723 ↛ 724line 723 didn't jump to line 724 because the condition on line 723 was never true
724 warnings.warn(
725 f"Extra kwargs for '{getattr(target, '__name__', target)}': {sorted(extras)}",
726 RuntimeWarning,
727 stacklevel=2,
728 )
729 kwargs = {k: v for k, v in kwargs.items() if k not in extras}
731 try:
732 bound = signature.bind_partial(*args, **kwargs)
733 except TypeError as exc:
734 raise ConstructorError(
735 f"Constructor signature mismatch for '{getattr(target, '__name__', target)}': {exc}.",
736 file_path=file_path,
737 config_path=config_path,
738 ) from exc
740 missing = list()
741 for name, param in signature.parameters.items():
742 if param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD) or param.default is not param.empty:
743 continue
744 elif name not in bound.arguments: 744 ↛ 745line 744 didn't jump to line 745 because the condition on line 744 was never true
745 missing.append(name)
747 if missing: 747 ↛ 748line 747 didn't jump to line 748 because the condition on line 747 was never true
748 raise ConstructorError(
749 f"Missing required parameters: {missing}.",
750 file_path=file_path,
751 config_path=config_path,
752 )
753 return kwargs
756if __name__ == "__main__": 756 ↛ 757line 756 didn't jump to line 757 because the condition on line 756 was never true
757 pass