Coverage for sygaldry / artificery.py: 87%
283 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-01 00:01 -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): 368 ↛ 369line 368 didn't jump to line 369 because the condition on line 368 was never true
369 break
370 elif "_type" in current or "_func" in current:
371 break
372 elif segment not in current: 372 ↛ 380line 372 didn't jump to line 380 because the condition on line 372 was always true
373 raise ConfigReferenceError(
374 f"Reference target '{ref_key}' not found "
375 f"('{segment}' missing under '{checked}').",
376 file_path=self._source_label,
377 config_path=ref_path,
378 )
379 else:
380 current = current[segment]
381 checked = f"{checked}.{segment}"
383 def _resolve_value(self, value: Any, *, path: list[str]) -> Any:
384 """
385 Resolve any value recursively.
387 :param value: Value to resolve.
388 :param path: Dotted path context.
389 :type value: object
390 :type path: list[str]
391 :returns: Resolved value.
392 :rtype: object
393 """
394 if isinstance(value, dict):
395 return self._resolve_dict(value, path=path)
396 elif isinstance(value, list):
397 return [
398 self._resolve_value(item, path=path + [str(idx)])
399 for idx, item in enumerate(value)
400 ]
401 elif isinstance(value, tuple): 401 ↛ 402line 401 didn't jump to line 402 because the condition on line 401 was never true
402 return tuple(
403 self._resolve_value(item, path=path + [str(idx)])
404 for idx, item in enumerate(value)
405 )
406 else:
407 return value
409 def _resolve_dict(self, value: dict[str, Any], *, path: list[str]) -> Any:
410 """
411 Resolve a dict, applying reserved-key rules.
413 :param value: Mapping to resolve.
414 :param path: Dotted path context.
415 :type value: dict[str, Any]
416 :type path: list[str]
417 :returns: Resolved mapping or constructed object.
418 :rtype: object
419 """
420 if "_ref" in value:
421 return self._resolve_ref(value["_ref"], path)
422 elif "_type" in value:
423 return self._resolve_type(value, path)
424 elif "_func" in value:
425 return self._resolve_func(value, path)
426 elif value.get("_deep") is False:
427 return {k: v for k, v in value.items() if k != "_deep"}
428 elif "_entries" in value:
429 return self._resolve_entries(value, path)
430 else:
431 resolved: dict[str, Any] = dict()
432 for key, child in value.items():
433 resolved[key] = self._resolve_value(child, path=path + [str(key)])
434 return resolved
436 def _resolve_ref(self, ref: Any, path: list[str]) -> Any:
437 """
438 Resolve a ``_ref`` mapping.
440 :param ref: Reference value.
441 :param path: Dotted path context.
442 :type ref: object
443 :type path: list[str]
444 :returns: Resolved referenced object or attribute.
445 :rtype: object
446 :raises ConfigReferenceError: If the ref is invalid or missing.
447 """
448 if not isinstance(ref, str) or not ref: 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true
449 raise ConfigReferenceError(
450 "_ref must be a non-empty string.",
451 file_path=self._source_label,
452 config_path=".".join(path),
453 )
454 else:
455 top, _, remainder = ref.partition(".")
456 target = self._resolve_top_level(top, path)
457 if remainder:
458 for attr in remainder.split("."):
459 try:
460 target = getattr(target, attr)
461 except AttributeError as exc:
462 raise ConfigReferenceError(
463 f"Attribute '{attr}' not found on reference '{top}'.",
464 file_path=self._source_label,
465 config_path=".".join(path),
466 ) from exc
467 return target
469 def _resolve_top_level(self, key: str, path: list[str]) -> Any:
470 """
471 Resolve a top-level config entry, with cycle detection.
473 :param key: Top-level key to resolve.
474 :param path: Dotted path context.
475 :type key: str
476 :type path: list[str]
477 :returns: Resolved value.
478 :rtype: object
479 :raises CircularReferenceError: If a cycle is detected.
480 """
481 if key in self._resolved_top:
482 return self._resolved_top[key]
483 elif key in self._resolving_top:
484 chain = " -> ".join(self._resolving_top + [key])
485 raise CircularReferenceError(
486 f"Circular reference detected: {chain}",
487 file_path=self._source_label,
488 config_path=".".join(path),
489 )
490 elif key not in self._active_config: 490 ↛ 491line 490 didn't jump to line 491 because the condition on line 490 was never true
491 raise ConfigReferenceError(
492 f"Reference target '{key}' not found.",
493 file_path=self._source_label,
494 config_path=".".join(path),
495 )
496 else:
497 self._resolving_top.append(key)
498 resolved = self._resolve_value(self._active_config[key], path=[key])
499 self._resolving_top.pop()
500 self._resolved_top[key] = resolved
501 return resolved
503 def _resolve_entries(self, value: dict[str, Any], path: list[str]) -> dict[Any, Any]:
504 """
505 Resolve a mapping encoded via ``_entries``.
507 :param value: Mapping containing ``_entries`` list.
508 :param path: Dotted path context.
509 :type value: dict[str, Any]
510 :type path: list[str]
511 :returns: Resolved mapping.
512 :rtype: dict[Any, Any]
513 :raises ResolutionError: If entries are malformed.
514 """
515 entries = value.get("_entries")
516 if not isinstance(entries, list): 516 ↛ 517line 516 didn't jump to line 517 because the condition on line 516 was never true
517 raise ResolutionError(
518 "_entries must be a list.",
519 file_path=self._source_label,
520 config_path=".".join(path),
521 )
522 elif len(value.keys() - {"_entries"}) != 0: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 raise ResolutionError(
524 "_entries cannot be combined with other keys.",
525 file_path=self._source_label,
526 config_path=".".join(path),
527 )
528 else:
529 resolved: dict[Any, Any] = dict()
530 for idx, entry in enumerate(entries):
531 if not isinstance(entry, dict) or "_key" not in entry or "_value" not in entry: 531 ↛ 532line 531 didn't jump to line 532 because the condition on line 531 was never true
532 raise ResolutionError(
533 "Each _entries item must contain _key and _value.",
534 file_path=self._source_label,
535 config_path=".".join(path + [str(idx)]),
536 )
537 key = self._resolve_value(entry["_key"], path=path + [str(idx), "_key"])
538 val = self._resolve_value(entry["_value"], path=path + [str(idx), "_value"])
539 try:
540 hash(key)
541 except Exception as exc:
542 raise ResolutionError(
543 "Resolved _entries keys must be hashable.",
544 file_path=self._source_label,
545 config_path=".".join(path + [str(idx), "_key"]),
546 ) from exc
548 if key in resolved: 548 ↛ 549line 548 didn't jump to line 549 because the condition on line 548 was never true
549 raise ResolutionError(
550 f"Duplicate _entries key '{key}'.",
551 file_path=self._source_label,
552 config_path=".".join(path + [str(idx), "_key"]),
553 )
554 else:
555 resolved[key] = val
556 return resolved
558 def _resolve_func(self, value: dict[str, Any], path: list[str]) -> Any:
559 """
560 Resolve a ``_func`` mapping to a callable.
562 :param value: Mapping containing ``_func``.
563 :param path: Dotted path context.
564 :type value: dict[str, Any]
565 :type path: list[str]
566 :returns: Imported callable.
567 :rtype: object
568 :raises ValidationError: If extra keys accompany ``_func``.
569 """
570 extras = set(value) - {"_func"}
571 if extras:
572 raise ValidationError(
573 f"_func does not accept other keys, found: {sorted(extras)}.",
574 file_path=self._source_label,
575 config_path=".".join(path),
576 )
577 else:
578 func_path = value.get("_func")
579 try:
580 return import_dotted_path(
581 func_path,
582 file_path=self._source_label,
583 config_path=".".join(path),
584 )
585 except ImportResolutionError:
586 raise
588 def _resolve_type(self, value: dict[str, Any], path: list[str]) -> Any:
589 """
590 Resolve a ``_type`` mapping into an instance.
592 :param value: Mapping containing ``_type`` and constructor data.
593 :param path: Dotted path context.
594 :type value: dict[str, Any]
595 :type path: list[str]
596 :returns: Constructed instance (possibly cached).
597 :rtype: object
598 """
599 type_path = value.get("_type")
600 args = value.get("_args", [])
601 extra_kwargs = value.get("_kwargs", {})
602 instance = value.get("_instance")
603 if not isinstance(args, list): 603 ↛ 604line 603 didn't jump to line 604 because the condition on line 603 was never true
604 raise ResolutionError(
605 "_args must be a list.",
606 file_path=self._source_label,
607 config_path=".".join(path),
608 )
609 elif not isinstance(extra_kwargs, dict):
610 raise ResolutionError(
611 "_kwargs must be a mapping.",
612 file_path=self._source_label,
613 config_path=".".join(path),
614 )
615 else:
616 kwargs = {k: v for k, v in value.items() if k not in RESERVED_KEYS}
617 resolved_args = tuple(self._resolve_value(arg, path=path + ["_args"]) for arg in args)
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):
743 continue
744 elif param.default is not param.empty:
745 continue
746 elif name not in bound.arguments: 746 ↛ 747line 746 didn't jump to line 747 because the condition on line 746 was never true
747 missing.append(name)
749 if missing: 749 ↛ 750line 749 didn't jump to line 750 because the condition on line 749 was never true
750 raise ConstructorError(
751 f"Missing required parameters: {missing}.",
752 file_path=file_path,
753 config_path=config_path,
754 )
755 return kwargs
758if __name__ == "__main__": 758 ↛ 759line 758 didn't jump to line 759 because the condition on line 758 was never true
759 pass