Coverage for sygaldry / loader.py: 86%
270 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-15 18:33 -0400
1from __future__ import annotations
3__author__ = "Rohan B. Dalton"
5import os
6import re
7import tempfile
8import urllib.request
9from collections.abc import Iterable
10from pathlib import Path
11from typing import Any, Optional
13import yaml
15from .errors import (
16 CircularIncludeError,
17 CircularInterpolationError,
18 IncludeError,
19 InterpolationError,
20 ParseError,
21)
23try:
24 import tomllib # type: ignore
25except ImportError: # pragma: no cover - fallback only
26 import tomli as tomllib # type: ignore
28_MISSING = object()
30_INTEGER_RE = re.compile(r"^(?P<integer>[-+]?\d+)$")
31_FLOAT_RE = re.compile(r"^(?P<float>[-+]?\d*\.\d+(?:[eE][-+]?\d+)?)$")
32_SCI_INT_RE = re.compile(r"^(?P<sci_int>[-+]?\d+[eE][-+]?\d+)$")
35def _is_url(path: str | Path) -> bool:
36 """
37 Return True if *path* looks like an HTTP(S) URL.
38 """
39 if isinstance(path, str):
40 return path.startswith("https://") or path.startswith("http://")
41 return False
44def _maybe_download(path: str | Path) -> Path:
45 """
46 If *path* is an HTTP(S) URL, download it to a temporary file and return
47 the local path. Otherwise return *path* as a :class:`~pathlib.Path`.
49 The suffix of the URL (ignoring query string / fragment) is preserved so
50 that format detection continues to work.
52 :param path: Local file path or HTTP(S) URL.
53 :type path: str | pathlib.Path
54 :returns: Local filesystem path to the config content.
55 :rtype: pathlib.Path
56 :raises ParseError: If the download fails.
57 """
58 if not _is_url(path):
59 return Path(path)
61 url = str(path)
62 # Derive suffix from URL (strip query/fragment first).
63 clean = url.split("?", maxsplit=1)[0].split("#", maxsplit=1)[0]
64 dot = clean.rfind(".")
65 suffix = clean[dot:] if dot != -1 else ""
67 with urllib.request.urlopen(url) as response:
68 content = response.read()
70 tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
71 try:
72 tmp.write(content)
73 finally:
74 tmp.close()
75 return Path(tmp.name)
78def load_config(path: str | Path) -> dict[str, Any]:
79 """
80 Load and interpolate a config file.
82 :param path: Path to a YAML or TOML config file, or an HTTP(S) URL.
83 :type path: str | pathlib.Path
84 :returns: Merged, interpolated configuration mapping.
85 :rtype: dict[str, Any]
86 """
87 local = _maybe_download(path)
88 visited: set[Path] = set()
89 data = _load_with_includes(local, ancestors=[], visited=visited)
90 return _interpolate_config(data, file_path=str(path))
93def _load_with_includes(
94 path: Path, ancestors: list[Path], visited: set[Path]
95) -> dict[str, Any]:
96 """
97 Load a file and recursively apply includes.
99 Uses *ancestors* (the current include stack) for true cycle detection
100 and *visited* (a global set) for diamond-include deduplication.
102 :param path: Path to the config file.
103 :param ancestors: Current include stack from root to parent.
104 :param visited: Global set of already-loaded paths.
105 :type path: pathlib.Path
106 :type ancestors: list[pathlib.Path]
107 :type visited: set[pathlib.Path]
108 :returns: Merged configuration mapping.
109 :rtype: dict[str, Any]
110 :raises CircularIncludeError: If a circular include is detected.
111 """
112 path = path.expanduser().resolve()
113 if path in ancestors:
114 chain_display = " -> ".join(str(p) for p in ancestors + [path])
115 raise CircularIncludeError(
116 f"Circular include detected: {chain_display}", file_path=str(path)
117 )
119 if path in visited:
120 return dict()
122 visited.add(path)
123 raw = _load_file(path)
124 includes = raw.pop("_include", None)
125 merged: dict[str, Any] = dict()
126 child_ancestors = ancestors + [path]
128 if includes:
129 if not isinstance(includes, list): 129 ↛ 130line 129 didn't jump to line 130 because the condition on line 129 was never true
130 raise IncludeError("_include must be a list of file paths.", file_path=str(path))
131 for include in includes:
132 include_path = _resolve_include(path, include)
133 data = _load_with_includes(include_path, child_ancestors, visited)
134 merged = _deep_merge(merged, data)
136 raw = _expand_dotted_keys(raw)
137 merged = _deep_merge(merged, raw)
138 return merged
141def _resolve_include(base: Path, include: Any) -> Path:
142 """
143 Resolve an include entry relative to its base file.
145 If *include* is an HTTP(S) URL it is downloaded to a temporary file
146 first.
148 :param base: Path to the including file.
149 :param include: Include entry value (path or HTTP(S) URL).
150 :type base: pathlib.Path
151 :type include: object
152 :returns: Absolute path to the included file.
153 :rtype: pathlib.Path
154 :raises IncludeError: If the include value is invalid.
155 """
156 if not isinstance(include, str): 156 ↛ 157line 156 didn't jump to line 157 because the condition on line 156 was never true
157 raise IncludeError("Include paths must be strings.", file_path=str(base))
158 if _is_url(include):
159 return _maybe_download(include)
160 candidate = Path(include)
161 if not candidate.is_absolute(): 161 ↛ 163line 161 didn't jump to line 163 because the condition on line 161 was always true
162 candidate = base.parent / candidate
163 return candidate
166def _expand_dotted_keys(data: dict[str, Any]) -> dict[str, Any]:
167 """
168 Expand top-level dotted keys into nested dicts.
170 For example, ``{"a.b.c": 1}`` becomes ``{"a": {"b": {"c": 1}}}``.
172 :param data: Raw config mapping.
173 :type data: dict[str, Any]
174 :returns: Mapping with dotted keys expanded.
175 :rtype: dict[str, Any]
176 """
177 result: dict[str, Any] = dict()
178 for key, value in data.items():
179 if "." not in key:
180 result[key] = value
181 continue
182 segments = key.split(".")
183 nested: dict[str, Any] = dict()
184 current = nested
185 for segment in segments[:-1]:
186 current[segment] = dict()
187 current = current[segment]
188 current[segments[-1]] = value
189 result = _deep_merge(result, nested)
190 return result
193def _load_file(path: Path) -> dict[str, Any]:
194 """
195 Load a YAML or TOML file into a mapping.
197 :param path: Path to the file on disk.
198 :type path: pathlib.Path
199 :returns: Parsed configuration mapping.
200 :rtype: dict[str, Any]
201 :raises ParseError: If reading or parsing fails.
202 """
203 suffix = path.suffix.lower()
204 try:
205 content = path.read_text(encoding="utf-8")
206 except Exception as exc:
207 raise ParseError(f"Failed to read config file: {path}", file_path=str(path)) from exc
209 if suffix in {".yaml", ".yml"}:
210 data = yaml.safe_load(content) or dict()
211 elif suffix == ".toml": 211 ↛ 214line 211 didn't jump to line 214 because the condition on line 211 was always true
212 data = tomllib.loads(content) or dict()
213 else:
214 raise ParseError(f"Unsupported config format: {suffix}", file_path=str(path))
216 if not isinstance(data, dict): 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true
217 raise ParseError("Top-level config must be a mapping.", file_path=str(path))
218 return data
221def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
222 """
223 Deep-merge two mappings (dicts merge, lists/scalars replace).
225 :param base: Base mapping.
226 :param override: Override mapping.
227 :type base: dict[str, Any]
228 :type override: dict[str, Any]
229 :returns: Merged mapping.
230 :rtype: dict[str, Any]
231 """
232 result = dict(base)
233 for key, value in override.items():
234 if key in result and isinstance(result[key], dict) and isinstance(value, dict):
235 result[key] = _deep_merge(result[key], value)
236 else:
237 result[key] = value
238 return result
241def _interpolate_config(config: dict[str, Any], *, file_path: str | None) -> dict[str, Any]:
242 """
243 Apply string interpolation to a config mapping.
245 :param config: Raw configuration mapping.
246 :param file_path: Source config path for context.
247 :type config: dict[str, Any]
248 :type file_path: str | None
249 :returns: Interpolated mapping.
250 :rtype: dict[str, Any]
251 """
252 resolver = _InterpolationResolver(config, file_path=file_path)
253 return resolver.resolve_value(config, path=[])
256class _InterpolationResolver:
257 """
258 Interpolation resolver for config mappings.
259 """
261 def __init__(self, raw: dict[str, Any], *, file_path: str | None) -> None:
262 """
263 Initialize the resolver.
265 :param raw: Raw configuration mapping.
266 :param file_path: Source config path for context.
267 :type raw: dict[str, Any]
268 :type file_path: str | None
269 """
270 self._raw = raw
271 self._file_path = file_path
272 self._cache: dict[str, Any] = dict()
273 self._visiting: list[str] = list()
275 def resolve_value(self, value: Any, *, path: list[str]) -> Any:
276 """
277 Resolve a value with interpolation.
279 :param value: Value to resolve.
280 :param path: Dotted path context.
281 :type value: object
282 :type path: list[str]
283 :returns: Resolved value.
284 :rtype: object
285 """
286 if isinstance(value, dict):
287 resolved = self._resolve_dict(value, path)
288 return resolved
289 elif isinstance(value, list):
290 resolved = [
291 self.resolve_value(item, path=path + [str(idx)])
292 for idx, item in enumerate(value)
293 ]
294 return resolved
295 elif isinstance(value, tuple): 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
296 resolved = tuple(
297 self.resolve_value(item, path=path + [str(idx)])
298 for idx, item in enumerate(value)
299 )
300 return resolved
301 elif isinstance(value, str):
302 interpolated = self._interpolate_string(value, path)
303 return interpolated
304 else:
305 return value
307 def _resolve_dict(self, value: dict[str, Any], path: list[str]) -> dict[str, Any]:
308 """
309 Resolve a dict by interpolating keys and values.
311 :param value: Mapping to resolve.
312 :param path: Dotted path context.
313 :type value: dict[str, Any]
314 :type path: list[str]
315 :returns: Resolved mapping.
316 :rtype: dict[str, Any]
317 """
318 resolved: dict[str, Any] = dict()
319 for key, val in value.items():
320 new_key = (
321 self.resolve_value(key, path=path + ["<key>"]) if isinstance(key, str) else key
322 )
323 if isinstance(new_key, str): 323 ↛ 326line 323 didn't jump to line 326 because the condition on line 323 was always true
324 key_path = path + [new_key]
325 else:
326 key_path = path + ["<key>"]
327 new_val = self.resolve_value(val, path=key_path)
328 if new_key in resolved: 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true
329 raise InterpolationError(
330 f"Interpolated key collision for '{new_key}'.",
331 file_path=self._file_path,
332 config_path=".".join(path),
333 )
334 resolved[new_key] = new_val
335 return resolved
337 def _interpolate_string(self, value: str, path: list[str]) -> Any:
338 """
339 Interpolate a string with placeholders.
341 :param value: String value to interpolate.
342 :param path: Dotted path context.
343 :type value: str
344 :type path: list[str]
345 :returns: Interpolated string or inferred scalar.
346 :rtype: object
347 """
348 parts, has_token, has_text = self._split_interpolation(value, path)
349 if not has_token:
350 if not parts: 350 ↛ 351line 350 didn't jump to line 351 because the condition on line 350 was never true
351 return value
352 return "".join(payload for _, payload in parts)
354 if has_token and not has_text and len(parts) == 1 and parts[0][0] == "token":
355 resolved = self._resolve_placeholder(parts[0][1].strip(), path)
356 if isinstance(resolved, str): 356 ↛ 358line 356 didn't jump to line 358 because the condition on line 356 was always true
357 return _infer_scalar(resolved)
358 return resolved
360 combined_parts: list[str] = list()
361 for kind, payload in parts:
362 if kind == "text":
363 combined_parts.append(payload)
364 continue
365 resolved = self._resolve_placeholder(payload.strip(), path)
366 combined_parts.append(str(resolved))
367 return "".join(combined_parts)
369 def _split_interpolation(
370 self, value: str, path: list[str]
371 ) -> tuple[list[tuple[str, str]], bool, bool]:
372 """
373 Split a string into interpolation and text parts.
375 :param value: String value to split.
376 :param path: Dotted path context.
377 :type value: str
378 :type path: list[str]
379 :returns: tuple of parts list, has-token flag, and has-text flag.
380 :rtype: tuple[list[tuple[str, str]], bool, bool]
381 :raises InterpolationError: If an interpolation is unterminated.
382 """
383 parts: list[tuple[str, str]] = list()
384 idx = 0
385 has_token = has_text = False
386 while idx < len(value):
387 if value.startswith("$${", idx):
388 parts.append(("text", "${"))
389 has_text = True
390 idx += 3
391 continue
392 elif value.startswith("${", idx):
393 has_token = True
394 idx += 2
395 depth = 1
396 buffer: list[str] = list()
397 while idx < len(value): 397 ↛ 418line 397 didn't jump to line 418 because the condition on line 397 was always true
398 if value.startswith("$${", idx): 398 ↛ 399line 398 didn't jump to line 399 because the condition on line 398 was never true
399 buffer.append("${")
400 idx += 3
401 continue
402 elif value.startswith("${", idx):
403 depth += 1
404 buffer.append("${")
405 idx += 2
406 continue
407 elif value[idx] == "}":
408 depth -= 1
409 idx += 1
410 if depth == 0:
411 break
412 buffer.append("}")
413 continue
414 else:
415 buffer.append(value[idx])
416 idx += 1
418 if depth != 0: 418 ↛ 419line 418 didn't jump to line 419 because the condition on line 418 was never true
419 raise InterpolationError(
420 "Unterminated interpolation.",
421 file_path=self._file_path,
422 config_path=".".join(path),
423 )
424 else:
425 parts.append(("token", "".join(buffer)))
426 continue
427 else:
428 parts.append(("text", value[idx]))
429 has_text = True
430 idx += 1
432 return parts, has_token, has_text
434 def _resolve_placeholder(self, token: str, path: list[str]) -> Any:
435 """
436 Resolve a single interpolation token.
438 Resolution order:
439 1. Config path lookup (if the key exists in the config tree).
440 2. Environment variable lookup.
441 3. Default value (if provided via ``:-``).
443 :param token: Token inside ``${...}``.
444 :param path: Dotted path context.
445 :type token: str
446 :type path: list[str]
447 :returns: Resolved value for the token.
448 :rtype: object
449 :raises InterpolationError: If the token cannot be resolved.
450 """
451 if ":-" in token:
452 key, default = token.split(":-", 1)
453 key = key.strip()
454 default = default.strip()
455 else:
456 key = token.strip()
457 default = None
459 config_value = _get_by_path(self._raw, key)
460 if config_value is not _MISSING:
461 return self._resolve_config_path(key, path)
462 elif (env_value := os.environ.get(key)) is not None:
463 return env_value
464 elif default is not None:
465 return self._interpolate_string(default, path)
466 else:
467 raise InterpolationError(
468 f"Interpolation target '{key}' not found in config or environment.",
469 file_path=self._file_path,
470 config_path=".".join(path),
471 )
473 def _resolve_config_path(self, key: str, path: list[str]) -> Any:
474 """
475 Resolve a config path interpolation.
477 :param key: Dotted config path.
478 :param path: Dotted path context.
479 :type key: str
480 :type path: list[str]
481 :returns: Resolved config value.
482 :rtype: object
483 :raises InterpolationError: If the path is missing.
484 """
485 dotted = key
486 if dotted in self._cache: 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true
487 return self._cache[dotted]
488 elif dotted in self._visiting:
489 raise CircularInterpolationError(
490 f"Circular interpolation detected for '{dotted}'.",
491 file_path=self._file_path,
492 config_path=".".join(path),
493 )
495 raw_value = _get_by_path(self._raw, dotted)
496 if raw_value is _MISSING: 496 ↛ 497line 496 didn't jump to line 497 because the condition on line 496 was never true
497 raise InterpolationError(
498 f"Interpolation target '{dotted}' not found.",
499 file_path=self._file_path,
500 config_path=".".join(path),
501 )
502 else:
503 self._visiting.append(dotted)
504 resolved = self.resolve_value(raw_value, path=dotted.split("."))
505 self._visiting.pop()
506 self._cache[dotted] = resolved
507 return resolved
510def _get_by_path(data: Any, dotted: str) -> Any:
511 """
512 Resolve a dotted path into a nested structure.
514 :param data: Root data structure.
515 :param dotted: Dotted path to resolve.
516 :type data: object
517 :type dotted: str
518 :returns: Resolved value or ``_MISSING`` sentinel.
519 :rtype: object
520 """
521 current = data
522 if dotted == "": 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 return _MISSING
524 else:
525 for segment in dotted.split("."):
526 if isinstance(current, dict): 526 ↛ 531line 526 didn't jump to line 531 because the condition on line 526 was always true
527 if segment not in current:
528 return _MISSING
529 else:
530 current = current[segment]
531 elif isinstance(current, list):
532 if not segment.isdigit():
533 return _MISSING
534 else:
535 idx = int(segment)
536 if idx < 0 or idx >= len(current):
537 return _MISSING
538 else:
539 current = current[idx]
540 else:
541 return _MISSING
542 return current
545def _infer_scalar(value: str) -> Any:
546 """
547 Infer a scalar type from a string.
549 :param value: String value to coerce.
550 :type value: str
551 :returns: Inferred scalar type or original string.
552 :rtype: object
553 """
554 lowered = value.strip().lower()
555 if lowered in {"true", "false"}:
556 return lowered == "true"
557 elif lowered in {"null", "none"}:
558 return None
560 stripped = value.strip()
561 if _INTEGER_RE.match(stripped):
562 func = int
563 elif _FLOAT_RE.match(stripped) or _SCI_INT_RE.match(stripped):
564 func = float
565 else:
566 func = None
568 if func is not None:
569 try:
570 return func(value)
571 except (TypeError, ValueError):
572 return value
573 else:
574 return value
577if __name__ == "__main__": 577 ↛ 578line 577 didn't jump to line 578 because the condition on line 577 was never true
578 pass