Coverage for sygaldry/loader.py: 85%

1from __future__ import annotations

3__author__ = "Rohan B. Dalton"

5import os

6import re

7from collections.abc import Iterable

8from pathlib import Path

9from typing import Any, Optional

11import yaml

13from .errors import (

14 CircularIncludeError,

15 CircularInterpolationError,

16 IncludeError,

17 InterpolationError,

18 ParseError,

19)

21try:

22 import tomllib # type: ignore

23except ImportError: # pragma: no cover - fallback only

24 import tomli as tomllib # type: ignore

26_MISSING = object()

28_INTEGER_RE = re.compile(r"^(?P<integer>[-+]?\d+)$")

29_FLOAT_RE = re.compile(r"^(?P<float>[-+]?\d*\.\d+(?:[eE][-+]?\d+)?)$")

30_SCI_INT_RE = re.compile(r"^(?P<sci_int>[-+]?\d+[eE][-+]?\d+)$")

33def load_config(path: Path) -> dict[str, Any]:

34 """

35 Load and interpolate a config file.

37 :param path: Path to a YAML or TOML config file.

38 :type path: pathlib.Path

39 :returns: Merged, interpolated configuration mapping.

40 :rtype: dict[str, Any]

41 """

42 visited: set[Path] = set()

43 data = _load_with_includes(path, ancestors=[], visited=visited)

44 return _interpolate_config(data, file_path=str(path))

47def _load_with_includes(

48 path: Path, ancestors: list[Path], visited: set[Path]

49) -> dict[str, Any]:

50 """

51 Load a file and recursively apply includes.

53 Uses *ancestors* (the current include stack) for true cycle detection

54 and *visited* (a global set) for diamond-include deduplication.

56 :param path: Path to the config file.

57 :param ancestors: Current include stack from root to parent.

58 :param visited: Global set of already-loaded paths.

59 :type path: pathlib.Path

60 :type ancestors: list[pathlib.Path]

61 :type visited: set[pathlib.Path]

62 :returns: Merged configuration mapping.

63 :rtype: dict[str, Any]

64 :raises CircularIncludeError: If a circular include is detected.

65 """

66 path = path.expanduser().resolve()

67 if path in ancestors:

68 chain_display = " -> ".join(str(p) for p in ancestors + [path])

69 raise CircularIncludeError(

70 f"Circular include detected: {chain_display}", file_path=str(path)

71 )

73 if path in visited:

74 return dict()

76 visited.add(path)

77 raw = _load_file(path)

78 includes = raw.pop("_include", None)

79 merged: dict[str, Any] = dict()

80 child_ancestors = ancestors + [path]

82 if includes:

83 if not isinstance(includes, list): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true

84 raise IncludeError("_include must be a list of file paths.", file_path=str(path))

85 for include in includes:

86 include_path = _resolve_include(path, include)

87 data = _load_with_includes(include_path, child_ancestors, visited)

88 merged = _deep_merge(merged, data)

90 raw = _expand_dotted_keys(raw)

91 merged = _deep_merge(merged, raw)

92 return merged

95def _resolve_include(base: Path, include: Any) -> Path:

96 """

97 Resolve an include entry relative to its base file.

99 :param base: Path to the including file.

100 :param include: Include entry value.

101 :type base: pathlib.Path

102 :type include: object

103 :returns: Absolute path to the included file.

104 :rtype: pathlib.Path

105 :raises IncludeError: If the include value is invalid.

106 """

107 if not isinstance(include, str): 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true

108 raise IncludeError("Include paths must be strings.", file_path=str(base))

109 candidate = Path(include)

110 if not candidate.is_absolute(): 110 ↛ 112line 110 didn't jump to line 112 because the condition on line 110 was always true

111 candidate = base.parent / candidate

112 return candidate

113

114

115def _expand_dotted_keys(data: dict[str, Any]) -> dict[str, Any]:

116 """

117 Expand top-level dotted keys into nested dicts.

118

119 For example, ``{"a.b.c": 1}`` becomes ``{"a": {"b": {"c": 1}}}``.

120

121 :param data: Raw config mapping.

122 :type data: dict[str, Any]

123 :returns: Mapping with dotted keys expanded.

124 :rtype: dict[str, Any]

125 """

126 result: dict[str, Any] = dict()

127 for key, value in data.items():

128 if "." not in key:

129 result[key] = value

130 continue

131 segments = key.split(".")

132 nested: dict[str, Any] = dict()

133 current = nested

134 for segment in segments[:-1]:

135 current[segment] = dict()

136 current = current[segment]

137 current[segments[-1]] = value

138 result = _deep_merge(result, nested)

139 return result

140

141

142def _load_file(path: Path) -> dict[str, Any]:

143 """

144 Load a YAML or TOML file into a mapping.

145

146 :param path: Path to the file on disk.

147 :type path: pathlib.Path

148 :returns: Parsed configuration mapping.

149 :rtype: dict[str, Any]

150 :raises ParseError: If reading or parsing fails.

151 """

152 suffix = path.suffix.lower()

153 try:

154 content = path.read_text(encoding="utf-8")

155 except Exception as exc:

156 raise ParseError(f"Failed to read config file: {path}", file_path=str(path)) from exc

157

158 if suffix in {".yaml", ".yml"}:

159 data = yaml.safe_load(content) or dict()

160 elif suffix == ".toml": 160 ↛ 163line 160 didn't jump to line 163 because the condition on line 160 was always true

161 data = tomllib.loads(content) or dict()

162 else:

163 raise ParseError(f"Unsupported config format: {suffix}", file_path=str(path))

164

165 if not isinstance(data, dict): 165 ↛ 166line 165 didn't jump to line 166 because the condition on line 165 was never true

166 raise ParseError("Top-level config must be a mapping.", file_path=str(path))

167 return data

168

169

170def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:

171 """

172 Deep-merge two mappings (dicts merge, lists/scalars replace).

173

174 :param base: Base mapping.

175 :param override: Override mapping.

176 :type base: dict[str, Any]

177 :type override: dict[str, Any]

178 :returns: Merged mapping.

179 :rtype: dict[str, Any]

180 """

181 result = dict(base)

182 for key, value in override.items():

183 if key in result and isinstance(result[key], dict) and isinstance(value, dict):

184 result[key] = _deep_merge(result[key], value)

185 else:

186 result[key] = value

187 return result

188

189

190def _interpolate_config(config: dict[str, Any], *, file_path: str | None) -> dict[str, Any]:

191 """

192 Apply string interpolation to a config mapping.

193

194 :param config: Raw configuration mapping.

195 :param file_path: Source config path for context.

196 :type config: dict[str, Any]

197 :type file_path: str | None

198 :returns: Interpolated mapping.

199 :rtype: dict[str, Any]

200 """

201 resolver = _InterpolationResolver(config, file_path=file_path)

202 return resolver.resolve_value(config, path=[])

203

204

205class _InterpolationResolver:

206 """

207 Interpolation resolver for config mappings.

208 """

209

210 def __init__(self, raw: dict[str, Any], *, file_path: str | None) -> None:

211 """

212 Initialize the resolver.

213

214 :param raw: Raw configuration mapping.

215 :param file_path: Source config path for context.

216 :type raw: dict[str, Any]

217 :type file_path: str | None

218 """

219 self._raw = raw

220 self._file_path = file_path

221 self._cache: dict[str, Any] = dict()

222 self._visiting: list[str] = list()

223

224 def resolve_value(self, value: Any, *, path: list[str]) -> Any:

225 """

226 Resolve a value with interpolation.

227

228 :param value: Value to resolve.

229 :param path: Dotted path context.

230 :type value: object

231 :type path: list[str]

232 :returns: Resolved value.

233 :rtype: object

234 """

235 if isinstance(value, dict):

236 resolved = self._resolve_dict(value, path)

237 return resolved

238 elif isinstance(value, list):

239 resolved = [

240 self.resolve_value(item, path=path + [str(idx)])

241 for idx, item in enumerate(value)

242 ]

243 return resolved

244 elif isinstance(value, tuple): 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 resolved = tuple(

246 self.resolve_value(item, path=path + [str(idx)])

247 for idx, item in enumerate(value)

248 )

249 return resolved

250 elif isinstance(value, str):

251 interpolated = self._interpolate_string(value, path)

252 return interpolated

253 else:

254 return value

255

256 def _resolve_dict(self, value: dict[str, Any], path: list[str]) -> dict[str, Any]:

257 """

258 Resolve a dict by interpolating keys and values.

259

260 :param value: Mapping to resolve.

261 :param path: Dotted path context.

262 :type value: dict[str, Any]

263 :type path: list[str]

264 :returns: Resolved mapping.

265 :rtype: dict[str, Any]

266 """

267 resolved: dict[str, Any] = dict()

268 for key, val in value.items():

269 new_key = (

270 self.resolve_value(key, path=path + ["<key>"]) if isinstance(key, str) else key

271 )

272 if isinstance(new_key, str): 272 ↛ 275line 272 didn't jump to line 275 because the condition on line 272 was always true

273 key_path = path + [new_key]

274 else:

275 key_path = path + ["<key>"]

276 new_val = self.resolve_value(val, path=key_path)

277 if new_key in resolved: 277 ↛ 278line 277 didn't jump to line 278 because the condition on line 277 was never true

278 raise InterpolationError(

279 f"Interpolated key collision for '{new_key}'.",

280 file_path=self._file_path,

281 config_path=".".join(path),

282 )

283 resolved[new_key] = new_val

284 return resolved

285

286 def _interpolate_string(self, value: str, path: list[str]) -> Any:

287 """

288 Interpolate a string with placeholders.

289

290 :param value: String value to interpolate.

291 :param path: Dotted path context.

292 :type value: str

293 :type path: list[str]

294 :returns: Interpolated string or inferred scalar.

295 :rtype: object

296 """

297 parts, has_token, has_text = self._split_interpolation(value, path)

298 if not has_token:

299 if not parts: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true

300 return value

301 return "".join(payload for _, payload in parts)

302

303 if has_token and not has_text and len(parts) == 1 and parts[0][0] == "token":

304 resolved = self._resolve_placeholder(parts[0][1].strip(), path)

305 if isinstance(resolved, str): 305 ↛ 307line 305 didn't jump to line 307 because the condition on line 305 was always true

306 return _infer_scalar(resolved)

307 return resolved

308

309 combined_parts: list[str] = list()

310 for kind, payload in parts:

311 if kind == "text":

312 combined_parts.append(payload)

313 continue

314 resolved = self._resolve_placeholder(payload.strip(), path)

315 combined_parts.append(str(resolved))

316 return "".join(combined_parts)

317

318 def _split_interpolation(

319 self, value: str, path: list[str]

320 ) -> tuple[list[tuple[str, str]], bool, bool]:

321 """

322 Split a string into interpolation and text parts.

323

324 :param value: String value to split.

325 :param path: Dotted path context.

326 :type value: str

327 :type path: list[str]

328 :returns: tuple of parts list, has-token flag, and has-text flag.

329 :rtype: tuple[list[tuple[str, str]], bool, bool]

330 :raises InterpolationError: If an interpolation is unterminated.

331 """

332 parts: list[tuple[str, str]] = list()

333 idx = 0

334 has_token = has_text = False

335 while idx < len(value):

336 if value.startswith("$${", idx):

337 parts.append(("text", "${"))

338 has_text = True

339 idx += 3

340 continue

341 elif value.startswith("${", idx):

342 has_token = True

343 idx += 2

344 depth = 1

345 buffer: list[str] = list()

346 while idx < len(value): 346 ↛ 367line 346 didn't jump to line 367 because the condition on line 346 was always true

347 if value.startswith("$${", idx): 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true

348 buffer.append("${")

349 idx += 3

350 continue

351 elif value.startswith("${", idx):

352 depth += 1

353 buffer.append("${")

354 idx += 2

355 continue

356 elif value[idx] == "}":

357 depth -= 1

358 idx += 1

359 if depth == 0:

360 break

361 buffer.append("}")

362 continue

363 else:

364 buffer.append(value[idx])

365 idx += 1

366

367 if depth != 0: 367 ↛ 368line 367 didn't jump to line 368 because the condition on line 367 was never true

368 raise InterpolationError(

369 "Unterminated interpolation.",

370 file_path=self._file_path,

371 config_path=".".join(path),

372 )

373 else:

374 parts.append(("token", "".join(buffer)))

375 continue

376 else:

377 parts.append(("text", value[idx]))

378 has_text = True

379 idx += 1

380

381 return parts, has_token, has_text

382

383 def _resolve_placeholder(self, token: str, path: list[str]) -> Any:

384 """

385 Resolve a single interpolation token.

386

387 Resolution order:

388 1. Config path lookup (if the key exists in the config tree).

389 2. Environment variable lookup.

390 3. Default value (if provided via ``:-``).

391

392 :param token: Token inside ``${...}``.

393 :param path: Dotted path context.

394 :type token: str

395 :type path: list[str]

396 :returns: Resolved value for the token.

397 :rtype: object

398 :raises InterpolationError: If the token cannot be resolved.

399 """

400 if ":-" in token:

401 key, default = token.split(":-", 1)

402 key = key.strip()

403 default = default.strip()

404 else:

405 key = token.strip()

406 default = None

407

408 config_value = _get_by_path(self._raw, key)

409 if config_value is not _MISSING:

410 return self._resolve_config_path(key, path)

411 else:

412 if (env_value := os.environ.get(key)) is not None:

413 return env_value

414 elif default is not None:

415 return self._interpolate_string(default, path)

416 else:

417 raise InterpolationError(

418 f"Interpolation target '{key}' not found in config or environment.",

419 file_path=self._file_path,

420 config_path=".".join(path),

421 )

422

423 def _resolve_config_path(self, key: str, path: list[str]) -> Any:

424 """

425 Resolve a config path interpolation.

426

427 :param key: Dotted config path.

428 :param path: Dotted path context.

429 :type key: str

430 :type path: list[str]

431 :returns: Resolved config value.

432 :rtype: object

433 :raises InterpolationError: If the path is missing.

434 """

435 dotted = key

436 if dotted in self._cache: 436 ↛ 437line 436 didn't jump to line 437 because the condition on line 436 was never true

437 return self._cache[dotted]

438 elif dotted in self._visiting:

439 raise CircularInterpolationError(

440 f"Circular interpolation detected for '{dotted}'.",

441 file_path=self._file_path,

442 config_path=".".join(path),

443 )

444

445 raw_value = _get_by_path(self._raw, dotted)

446 if raw_value is _MISSING: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true

447 raise InterpolationError(

448 f"Interpolation target '{dotted}' not found.",

449 file_path=self._file_path,

450 config_path=".".join(path),

451 )

452 else:

453 self._visiting.append(dotted)

454 resolved = self.resolve_value(raw_value, path=dotted.split("."))

455 self._visiting.pop()

456 self._cache[dotted] = resolved

457 return resolved

458

459

460def _get_by_path(data: Any, dotted: str) -> Any:

461 """

462 Resolve a dotted path into a nested structure.

463

464 :param data: Root data structure.

465 :param dotted: Dotted path to resolve.

466 :type data: object

467 :type dotted: str

468 :returns: Resolved value or ``_MISSING`` sentinel.

469 :rtype: object

470 """

471 current = data

472 if dotted == "": 472 ↛ 473line 472 didn't jump to line 473 because the condition on line 472 was never true

473 return _MISSING

474 else:

475 for segment in dotted.split("."):

476 if isinstance(current, dict): 476 ↛ 481line 476 didn't jump to line 481 because the condition on line 476 was always true

477 if segment not in current:

478 return _MISSING

479 else:

480 current = current[segment]

481 elif isinstance(current, list):

482 if not segment.isdigit():

483 return _MISSING

484 else:

485 idx = int(segment)

486 if idx < 0 or idx >= len(current):

487 return _MISSING

488 else:

489 current = current[idx]

490 else:

491 return _MISSING

492 else:

493 return current

494

495

496def _infer_scalar(value: str) -> Any:

497 """

498 Infer a scalar type from a string.

499

500 :param value: String value to coerce.

501 :type value: str

502 :returns: Inferred scalar type or original string.

503 :rtype: object

504 """

505 lowered = value.strip().lower()

506 if lowered in {"true", "false"}:

507 return lowered == "true"

508 elif lowered in {"null", "none"}:

509 return None

510

511 stripped = value.strip()

512 if _INTEGER_RE.match(stripped):

513 func = int

514 elif _FLOAT_RE.match(stripped) or _SCI_INT_RE.match(stripped):

515 func = float

516 else:

517 func = None

518

519 if func is not None:

520 try:

521 return func(value)

522 except (TypeError, ValueError):

523 return value

524 else:

525 return value

526

527

528if __name__ == "__main__": 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true

529 pass

Coverage for sygaldry / loader.py: 85%

247 statements