Coverage for sygaldry / artificery.py: 88%
279 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-10 13:02 -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 _maybe_download,
28 load_config,
29)
30from .types import RESERVED_KEYS, import_dotted_path
33@dataclass(frozen=True)
34class Reference:
35 """
36 Reference to a dotted config path.
37 """
39 path: str
42@dataclass(frozen=True)
43class ConstructorSpec:
44 """
45 Constructor specification for a component.
46 """
48 target: str
49 args: tuple[Any, ...]
50 kwargs: dict[str, Any]
51 instance: str | None
54@dataclass(frozen=True)
55class Node:
56 """
57 Intermediate node representation for optional IR usage.
58 """
60 name: str
61 children: tuple[str, ...]
62 bindings: dict[str, Any]
65def _set_by_path(data: dict[str, Any], dotted_path: str, value: Any) -> None:
66 """
67 Set a nested value by dotted path, creating intermediate dicts.
69 :param data: Root mapping to modify in-place.
70 :param dotted_path: Dotted key path (e.g. ``"db.host"``).
71 :param value: Value to set.
72 :type data: dict
73 :type dotted_path: str
74 :type value: object
75 """
76 segments = dotted_path.split(".")
77 current = data
78 for segment in segments[:-1]:
79 if segment not in current or not isinstance(current[segment], dict):
80 current[segment] = dict()
81 current = current[segment]
82 current[segments[-1]] = value
85def load_config_file(path: str | Path) -> dict[str, Any]:
86 """
87 Load a config file without resolving components.
89 :param path: Path to a YAML or TOML config file, or an HTTP(S) URL.
90 :type path: str | pathlib.Path
91 :returns: Parsed and interpolated mapping.
92 :rtype: dict[str, Any]
93 """
94 return load_config(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, or an HTTP(S) URL.
107 :type path: str | pathlib.Path
108 """
109 self._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 if isinstance(path, str) else 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 local = _maybe_download(path)
250 resolved_path = local.expanduser().resolve()
251 visited: set[Path] = set()
252 data = _load_with_includes(resolved_path, ancestors=[], visited=visited)
253 merged = _deep_merge(merged, data)
254 return merged
256 def _apply_uses(self, config: dict[str, Any]) -> dict[str, Any]:
257 """
258 Copy values within the config for each use mapping.
260 :param config: Merged raw config.
261 :type config: dict[str, Any]
262 :returns: Modified config.
263 :rtype: dict[str, Any]
264 """
265 for target_path, source_path in self._uses.items():
266 value = _get_by_path(config, source_path)
267 if value is _MISSING:
268 available = sorted(config.keys())
269 raise ConfigReferenceError(
270 f"Use source '{source_path}' not found. "
271 f"Available top-level keys: {available}",
272 file_path=self._source_label,
273 config_path=target_path,
274 )
275 _set_by_path(config, target_path, value)
276 return config
278 def _apply_overrides(self, config: dict[str, Any]) -> dict[str, Any]:
279 """
280 Apply dotted-path overrides to the config.
282 :param config: Merged raw config.
283 :type config: dict[str, Any]
284 :returns: Modified config.
285 :rtype: dict[str, Any]
286 """
287 for key, value in self._overrides.items():
288 _set_by_path(config, key, value)
289 return config
291 def resolve(self) -> dict[str, Any]:
292 """
293 Validate and resolve the full config mapping into objects.
295 :returns: Resolved configuration mapping.
296 :rtype: dict[str, Any]
297 """
298 active = self.config
299 self._active_config = active
300 self._resolved_top: dict[str, Any] = dict()
301 self._resolving_top: list[str] = list()
302 self._validate_schema(active, path=[])
303 self._validate_refs(active)
304 return self._resolve_value(active, path=[])
306 def _validate_schema(self, value: Any, *, path: list[str]) -> None:
307 """
308 Validate schema rules for reserved keys.
310 :param value: Value to validate.
311 :param path: Dotted path context.
312 :type value: object
313 :type path: list[str]
314 :raises ValidationError: On schema violations.
315 """
316 if isinstance(value, dict):
317 if "_ref" in value and len(value) != 1:
318 raise ValidationError(
319 "_ref must be the only key in its mapping.",
320 file_path=self._source_label,
321 config_path=".".join(path),
322 )
323 elif "_type" in value and "_func" in value: 323 ↛ 324line 323 didn't jump to line 324 because the condition on line 323 was never true
324 raise ValidationError(
325 "_type and _func are mutually exclusive.",
326 file_path=self._source_label,
327 config_path=".".join(path),
328 )
329 elif "_instance" in value and "_type" not in value: 329 ↛ 330line 329 didn't jump to line 330 because the condition on line 329 was never true
330 raise ValidationError(
331 "_instance is only valid with _type.",
332 file_path=self._source_label,
333 config_path=".".join(path),
334 )
336 for key, child in value.items():
337 next_path = path + [str(key)]
338 self._validate_schema(child, path=next_path)
340 elif isinstance(value, list) or isinstance(value, tuple):
341 for idx, child in enumerate(value):
342 self._validate_schema(child, path=path + [str(idx)])
344 def _validate_refs(self, config: dict[str, Any]) -> None:
345 """
346 Validate that all _ref targets exist in the config tree.
348 For plain config paths (no ``_type`` at intermediate levels),
349 validates the full dotted path through the raw config dicts.
350 Stops validation at dict boundaries since deeper segments
351 may be attribute accesses on constructed objects.
353 :param config: Configuration mapping.
354 :type config: dict[str, Any]
355 :raises ConfigReferenceError: If any target is missing.
356 """
357 for ref_path, ref_key in _collect_refs(config):
358 top = ref_key.split(".", 1)[0]
359 if top not in config:
360 raise ConfigReferenceError(
361 f"Reference target '{ref_key}' not found.",
362 file_path=self._source_label,
363 config_path=ref_path,
364 )
365 current = config[top]
366 segments = ref_key.split(".")[1:]
367 checked = top
368 for segment in segments:
369 if not isinstance(current, dict) or "_type" in current or "_func" in current:
370 break
371 elif segment not in current: 371 ↛ 379line 371 didn't jump to line 379 because the condition on line 371 was always true
372 raise ConfigReferenceError(
373 f"Reference target '{ref_key}' not found "
374 f"('{segment}' missing under '{checked}').",
375 file_path=self._source_label,
376 config_path=ref_path,
377 )
378 else:
379 current = current[segment]
380 checked = f"{checked}.{segment}"
382 def _resolve_value(self, value: Any, *, path: list[str]) -> Any:
383 """
384 Resolve any value recursively.
386 :param value: Value to resolve.
387 :param path: Dotted path context.
388 :type value: object
389 :type path: list[str]
390 :returns: Resolved value.
391 :rtype: object
392 """
393 if isinstance(value, dict):
394 return self._resolve_dict(value, path=path)
395 elif isinstance(value, list):
396 return [
397 self._resolve_value(item, path=path + [str(idx)])
398 for idx, item in enumerate(value)
399 ]
400 elif isinstance(value, tuple): 400 ↛ 401line 400 didn't jump to line 401 because the condition on line 400 was never true
401 return tuple(
402 self._resolve_value(item, path=path + [str(idx)])
403 for idx, item in enumerate(value)
404 )
405 else:
406 return value
408 def _resolve_dict(self, value: dict[str, Any], *, path: list[str]) -> Any:
409 """
410 Resolve a dict, applying reserved-key rules.
412 :param value: Mapping to resolve.
413 :param path: Dotted path context.
414 :type value: dict[str, Any]
415 :type path: list[str]
416 :returns: Resolved mapping or constructed object.
417 :rtype: object
418 """
419 if "_ref" in value:
420 return self._resolve_ref(value["_ref"], path)
421 elif "_type" in value:
422 return self._resolve_type(value, path)
423 elif "_func" in value:
424 return self._resolve_func(value, path)
425 elif value.get("_deep") is False:
426 return {k: v for k, v in value.items() if k != "_deep"}
427 elif "_entries" in value:
428 return self._resolve_entries(value, path)
429 else:
430 resolved: dict[str, Any] = dict()
431 for key, child in value.items():
432 resolved[key] = self._resolve_value(child, path=path + [str(key)])
433 return resolved
435 def _resolve_ref(self, ref: Any, path: list[str]) -> Any:
436 """
437 Resolve a ``_ref`` mapping.
439 :param ref: Reference value.
440 :param path: Dotted path context.
441 :type ref: object
442 :type path: list[str]
443 :returns: Resolved referenced object or attribute.
444 :rtype: object
445 :raises ConfigReferenceError: If the ref is invalid or missing.
446 """
447 if not isinstance(ref, str) or not ref: 447 ↛ 448line 447 didn't jump to line 448 because the condition on line 447 was never true
448 raise ConfigReferenceError(
449 "_ref must be a non-empty string.",
450 file_path=self._source_label,
451 config_path=".".join(path),
452 )
453 else:
454 top, _, remainder = ref.partition(".")
455 target = self._resolve_top_level(top, path)
456 if remainder:
457 for attr in remainder.split("."):
458 try:
459 target = getattr(target, attr)
460 except AttributeError as exc:
461 raise ConfigReferenceError(
462 f"Attribute '{attr}' not found on reference '{top}'.",
463 file_path=self._source_label,
464 config_path=".".join(path),
465 ) from exc
466 return target
468 def _resolve_top_level(self, key: str, path: list[str]) -> Any:
469 """
470 Resolve a top-level config entry, with cycle detection.
472 :param key: Top-level key to resolve.
473 :param path: Dotted path context.
474 :type key: str
475 :type path: list[str]
476 :returns: Resolved value.
477 :rtype: object
478 :raises CircularReferenceError: If a cycle is detected.
479 """
480 if key in self._resolved_top:
481 return self._resolved_top[key]
482 elif key in self._resolving_top:
483 chain = " -> ".join(self._resolving_top + [key])
484 raise CircularReferenceError(
485 f"Circular reference detected: {chain}",
486 file_path=self._source_label,
487 config_path=".".join(path),
488 )
489 elif key not in self._active_config: 489 ↛ 490line 489 didn't jump to line 490 because the condition on line 489 was never true
490 raise ConfigReferenceError(
491 f"Reference target '{key}' not found.",
492 file_path=self._source_label,
493 config_path=".".join(path),
494 )
495 else:
496 self._resolving_top.append(key)
497 resolved = self._resolve_value(self._active_config[key], path=[key])
498 self._resolving_top.pop()
499 self._resolved_top[key] = resolved
500 return resolved
502 def _resolve_entries(self, value: dict[str, Any], path: list[str]) -> dict[Any, Any]:
503 """
504 Resolve a mapping encoded via ``_entries``.
506 :param value: Mapping containing ``_entries`` list.
507 :param path: Dotted path context.
508 :type value: dict[str, Any]
509 :type path: list[str]
510 :returns: Resolved mapping.
511 :rtype: dict[Any, Any]
512 :raises ResolutionError: If entries are malformed.
513 """
514 entries = value.get("_entries")
515 if not isinstance(entries, list): 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 raise ResolutionError(
517 "_entries must be a list.",
518 file_path=self._source_label,
519 config_path=".".join(path),
520 )
521 elif len(value.keys() - {"_entries"}) != 0: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 raise ResolutionError(
523 "_entries cannot be combined with other keys.",
524 file_path=self._source_label,
525 config_path=".".join(path),
526 )
527 else:
528 resolved: dict[Any, Any] = dict()
529 for idx, entry in enumerate(entries):
530 if not isinstance(entry, dict) or "_key" not in entry or "_value" not in entry: 530 ↛ 531line 530 didn't jump to line 531 because the condition on line 530 was never true
531 raise ResolutionError(
532 "Each _entries item must contain _key and _value.",
533 file_path=self._source_label,
534 config_path=".".join(path + [str(idx)]),
535 )
536 key = self._resolve_value(entry["_key"], path=path + [str(idx), "_key"])
537 val = self._resolve_value(entry["_value"], path=path + [str(idx), "_value"])
538 try:
539 hash(key)
540 except Exception as exc:
541 raise ResolutionError(
542 "Resolved _entries keys must be hashable.",
543 file_path=self._source_label,
544 config_path=".".join(path + [str(idx), "_key"]),
545 ) from exc
547 if key in resolved: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true
548 raise ResolutionError(
549 f"Duplicate _entries key '{key}'.",
550 file_path=self._source_label,
551 config_path=".".join(path + [str(idx), "_key"]),
552 )
553 else:
554 resolved[key] = val
555 return resolved
557 def _resolve_func(self, value: dict[str, Any], path: list[str]) -> Any:
558 """
559 Resolve a ``_func`` mapping to a callable.
561 :param value: Mapping containing ``_func``.
562 :param path: Dotted path context.
563 :type value: dict[str, Any]
564 :type path: list[str]
565 :returns: Imported callable.
566 :rtype: object
567 :raises ValidationError: If extra keys accompany ``_func``.
568 """
569 extras = set(value) - {"_func"}
570 if extras:
571 raise ValidationError(
572 f"_func does not accept other keys, found: {sorted(extras)}.",
573 file_path=self._source_label,
574 config_path=".".join(path),
575 )
576 else:
577 func_path = value.get("_func")
578 try:
579 return import_dotted_path(
580 func_path,
581 file_path=self._source_label,
582 config_path=".".join(path),
583 )
584 except ImportResolutionError:
585 raise
587 def _resolve_type(self, value: dict[str, Any], path: list[str]) -> Any:
588 """
589 Resolve a ``_type`` mapping into an instance.
591 :param value: Mapping containing ``_type`` and constructor data.
592 :param path: Dotted path context.
593 :type value: dict[str, Any]
594 :type path: list[str]
595 :returns: Constructed instance (possibly cached).
596 :rtype: object
597 """
598 type_path = value.get("_type")
599 args = value.get("_args", [])
600 extra_kwargs = value.get("_kwargs", {})
601 instance = value.get("_instance")
602 if not isinstance(args, list): 602 ↛ 603line 602 didn't jump to line 603 because the condition on line 602 was never true
603 raise ResolutionError(
604 "_args must be a list.",
605 file_path=self._source_label,
606 config_path=".".join(path),
607 )
608 elif not isinstance(extra_kwargs, dict):
609 raise ResolutionError(
610 "_kwargs must be a mapping.",
611 file_path=self._source_label,
612 config_path=".".join(path),
613 )
614 else:
615 kwargs = {k: v for k, v in value.items() if k not in RESERVED_KEYS}
616 resolved_args = tuple(
617 self._resolve_value(arg, path=path + ["_args"]) for arg in args
618 )
619 resolved_kwargs = {
620 key: self._resolve_value(val, path=path + [key]) for key, val in kwargs.items()
621 }
622 for key, val in extra_kwargs.items():
623 resolved_kwargs[key] = self._resolve_value(val, path=path + ["_kwargs", key])
625 target = import_dotted_path(
626 type_path,
627 file_path=self._source_label,
628 config_path=".".join(path),
629 )
631 resolved_kwargs = _validate_signature(
632 target,
633 resolved_args,
634 resolved_kwargs,
635 file_path=self._source_label,
636 config_path=".".join(path),
637 )
639 def factory() -> Any:
640 try:
641 return target(*resolved_args, **resolved_kwargs)
642 except Exception as exc: # noqa: BLE001
643 raise ConstructorError(
644 f"Failed to construct '{type_path}'.",
645 file_path=self._source_label,
646 config_path=".".join(path),
647 ) from exc
649 instance = self._cache.get_or_create(
650 type_path,
651 instance,
652 resolved_args,
653 resolved_kwargs,
654 factory,
655 transient=self._transient,
656 file_path=self._source_label,
657 config_path=".".join(path),
658 )
659 return instance
662def _collect_refs(value: Any, *, path: list[str] | None = None) -> Iterable[tuple[str, str]]:
663 """
664 Collect ``_ref`` occurrences from a config tree.
666 :param value: Root value to inspect.
667 :param path: Current path stack.
668 :type value: object
669 :type path: list[str] | None
670 :returns: Iterable of (config_path, ref_target) pairs.
671 :rtype: collections.abc.Iterable[tuple[str, str]]
672 """
673 path = path or list()
674 if isinstance(value, dict):
675 if "_ref" in value:
676 yield ".".join(path), value["_ref"]
677 return
678 else:
679 for key, child in value.items():
680 yield from _collect_refs(child, path=path + [str(key)])
681 elif isinstance(value, list) or isinstance(value, tuple):
682 for idx, child in enumerate(value):
683 yield from _collect_refs(child, path=path + [str(idx)])
686def _validate_signature(
687 target: Any,
688 args: tuple[Any, ...],
689 kwargs: dict[str, Any],
690 *,
691 file_path: str | None,
692 config_path: str,
693) -> dict[str, Any]:
694 """
695 Validate constructor signature and filter extra kwargs.
697 :param target: Callable to validate.
698 :param args: Positional arguments.
699 :param kwargs: Keyword arguments.
700 :param file_path: Source config path for context.
701 :param config_path: Dotted config path for context.
702 :type target: object
703 :type args: tuple[Any, ...]
704 :type kwargs: dict[str, Any]
705 :type file_path: str | None
706 :type config_path: str
707 :returns: Possibly filtered kwargs.
708 :rtype: dict[str, Any]
709 :raises ConstructorError: If required parameters are missing.
710 """
711 try:
712 signature = inspect.signature(target)
713 except (TypeError, ValueError):
714 return kwargs
716 has_var_kw = any(p.kind == p.VAR_KEYWORD for p in signature.parameters.values())
717 if not has_var_kw:
718 allowed = {
719 name
720 for name, param in signature.parameters.items()
721 if param.kind in (param.POSITIONAL_OR_KEYWORD, param.KEYWORD_ONLY)
722 }
723 extras = set(kwargs) - allowed
724 if extras: 724 ↛ 725line 724 didn't jump to line 725 because the condition on line 724 was never true
725 warnings.warn(
726 f"Extra kwargs for '{getattr(target, '__name__', target)}': {sorted(extras)}",
727 RuntimeWarning,
728 stacklevel=2,
729 )
730 kwargs = {k: v for k, v in kwargs.items() if k not in extras}
732 try:
733 bound = signature.bind_partial(*args, **kwargs)
734 except TypeError as exc:
735 raise ConstructorError(
736 f"Constructor signature mismatch for '{getattr(target, '__name__', target)}': {exc}.",
737 file_path=file_path,
738 config_path=config_path,
739 ) from exc
741 missing = list()
742 for name, param in signature.parameters.items():
743 if (
744 param.kind in (param.VAR_POSITIONAL, param.VAR_KEYWORD)
745 or param.default is not param.empty
746 ):
747 continue
748 elif name not in bound.arguments: 748 ↛ 749line 748 didn't jump to line 749 because the condition on line 748 was never true
749 missing.append(name)
751 if missing: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true
752 raise ConstructorError(
753 f"Missing required parameters: {missing}.",
754 file_path=file_path,
755 config_path=config_path,
756 )
757 return kwargs
760if __name__ == "__main__": 760 ↛ 761line 760 didn't jump to line 761 because the condition on line 760 was never true
761 pass