Coverage for pyTooling/Configuration/YAML.py: 96%

1# ==================================================================================================================== # 

2# _____ _ _ ____ __ _ _ _ # 

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ / ___|___ _ __ / _(_) __ _ _ _ _ __ __ _| |_(_) ___ _ __ # 

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` || | / _ \| '_ \| |_| |/ _` | | | | '__/ _` | __| |/ _ \| '_ \ # 

5# | |_) | |_| || | (_) | (_) | | | | | | (_| || |__| (_) | | | | _| | (_| | |_| | | | (_| | |_| | (_) | | | | # 

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____\___/|_| |_|_| |_|\__, |\__,_|_| \__,_|\__|_|\___/|_| |_| # 

7# |_| |___/ |___/ |___/ # 

8# ==================================================================================================================== # 

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

13# ==================================================================================================================== # 

14# Copyright 2021-2026 Patrick Lehmann - Bötzingen, Germany # 

15# # 

16# Licensed under the Apache License, Version 2.0 (the "License"); # 

17# you may not use this file except in compliance with the License. # 

18# You may obtain a copy of the License at # 

19# # 

20# http://www.apache.org/licenses/LICENSE-2.0 # 

21# # 

22# Unless required by applicable law or agreed to in writing, software # 

23# distributed under the License is distributed on an "AS IS" BASIS, # 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

25# See the License for the specific language governing permissions and # 

26# limitations under the License. # 

27# # 

28# SPDX-License-Identifier: Apache-2.0 # 

29# ==================================================================================================================== # 

30# 

31""" 

32Configuration reader for YAML files. 

33 

34.. hint:: 

35 

36 See :ref:`high-level help <CONFIG/FileFormat/YAML>` for explanations and usage examples. 

37""" 

38from __future__ import annotations 

39 

40from pathlib import Path 

41from typing import Any, Union, Iterator as typing_Iterator, Self 

42 

43from pyTooling.Exceptions import MissingDependencyException 

44 

45try: 

46 from ruamel.yaml import YAML, CommentedMap, CommentedSeq 

47except ImportError as ex: # pragma: no cover 

48 raise MissingDependencyException(dependency="ruamel.yaml", extra="yaml") from ex 

49 

50from pyTooling.Common import getFullyQualifiedName 

51from pyTooling.Decorators import export, InheritDocString 

52from pyTooling.MetaClasses import ExtendedType 

53from pyTooling.Configuration import ConfigurationException, KeyT, NodeT, ValueT 

54from pyTooling.Configuration import InterpolationException, KeyNotFoundException, PathExpressionException 

55from pyTooling.Configuration import UnsupportedValueTypeException 

56from pyTooling.Configuration import Node as Abstract_Node 

57from pyTooling.Configuration import Dictionary as Abstract_Dict 

58from pyTooling.Configuration import Sequence as Abstract_Seq 

59from pyTooling.Configuration import Configuration as Abstract_Configuration 

60 

61 

62@export 

63class Node(Abstract_Node): 

64 """ 

65 Node in a YAML configuration data structure. 

66 """ 

67 

68 _yamlNode: Union[CommentedMap, CommentedSeq] #: Reference to the associated YAML node. 

69 _cache: dict[str, ValueT] #: Cache of already converted sub-nodes and values, by key. 

70 _key: KeyT #: Key of this node. 

71 _length: int #: Number of sub-elements. 

72 

73 def __init__( 

74 self, 

75 root: Configuration, 

76 parent: NodeT, 

77 key: KeyT, 

78 yamlNode: Union[CommentedMap, CommentedSeq] 

79 ) -> None: 

80 """ 

81 Initializes a YAML node. 

82 

83 :param root: Reference to the root node. 

84 :param parent: Reference to the parent node. 

85 :param key: Key of the node within its parent. 

86 :param yamlNode: Reference to the YAML node. 

87 """ 

88 Abstract_Node.__init__(self, root, parent) 

89 

90 self._yamlNode = yamlNode 

91 self._cache = {} 

92 self._key = key 

93 self._length = len(yamlNode) 

94 

95 @InheritDocString(Abstract_Node) 

96 def __len__(self) -> int: 

97 return self._length 

98 

99 @InheritDocString(Abstract_Node) 

100 def __getitem__(self, key: KeyT) -> ValueT: 

101 return self._GetNodeOrValue(str(key)) 

102 

103 @property 

104 def Key(self) -> KeyT: 

105 """ 

106 Property to access the node's key. 

107 

108 :returns: Key of the node. 

109 :raises NotImplementedError: If a new key is assigned; renaming a key is not supported by this configuration 

110 implementation. 

111 """ 

112 return self._key 

113 

114 @Key.setter 

115 def Key(self, value: KeyT) -> None: 

116 raise NotImplementedError() 

117 

118 @InheritDocString(Abstract_Node) 

119 def QueryPath(self, query: str) -> ValueT: 

120 path = self._ToPath(query) 

121 return self._GetNodeOrValueByPathExpression(path) 

122 

123 @staticmethod 

124 def _ToPath(query: str) -> list[Union[str, int]]: 

125 """ 

126 Split a path expression into its elements. 

127 

128 :param query: Path expression, with its elements separated by ``:``. 

129 :returns: List of keys and indices. 

130 """ 

131 return query.split(":") 

132 

133 def _LookupKey(self, key: str) -> Any: 

134 """ 

135 Look up a key in the YAML node, trying it as string, integer and float. 

136 

137 :param key: Key or index to look up. 

138 :returns: The raw value as returned by the YAML parser. 

139 :raises KeyNotFoundException: If the key exists neither as string, nor as integer or float. 

140 """ 

141 try: 

142 return self._yamlNode[key] 

143 except (KeyError, TypeError): 

144 pass 

145 

146 for conversion in (int, float): 

147 try: 

148 convertedKey = conversion(key) 

149 except ValueError: 

150 continue 

151 

152 try: 

153 return self._yamlNode[convertedKey] 

154 except (KeyError, IndexError, TypeError): 

155 pass 

156 

157 ex = KeyNotFoundException(f"Key '{key}' not found in node '{self._key}'.") 

158 ex.add_note(self._DescribeKeys()) 

159 raise ex 

160 

161 def _DescribeKeys(self) -> str: 

162 """ 

163 Describe the keys or indices offered by this node, so it can be used as an exception note. 

164 

165 :returns: A one-line description of the node's keys or index range. 

166 """ 

167 if isinstance(self._yamlNode, CommentedMap): 

168 if self._length == 0: 

169 return f"Node '{self._key}' is an empty dictionary." 

170 

171 keys = "', '".join(str(key) for key in self._yamlNode) 

172 return f"Available keys: '{keys}'." 

173 else: 

174 if self._length == 0: 

175 return f"Node '{self._key}' is an empty sequence." 

176 

177 return f"Node '{self._key}' is a sequence with indices 0..{self._length - 1}." 

178 

179 def _GetNodeOrValue(self, key: str) -> ValueT: 

180 """ 

181 Return a sub-node or a value by key, converting it on first access. 

182 

183 The converted object is cached, so a second access returns the same node object rather than a new one. 

184 

185 :param key: Key or index to look up. 

186 :returns: A dictionary node, a sequence node, or a scalar value with its variables 

187 resolved. 

188 :raises KeyNotFoundException: If the key doesn't exist in this node. 

189 :raises UnsupportedValueTypeException: If the YAML parser returned a value that is neither a scalar, nor a 

190 node. 

191 """ 

192 try: 

193 value = self._cache[key] 

194 except KeyError: 

195 value = self._LookupKey(key) 

196 

197 if isinstance(value, str): 

198 value = self._ResolveVariables(value) 

199 elif isinstance(value, (int, float)): 

200 value = str(value) 

201 elif isinstance(value, CommentedMap): 

202 value = self.DICT_TYPE(self, self, key, value) 

203 elif isinstance(value, CommentedSeq): 

204 value = self.SEQ_TYPE(self, self, key, value) 

205 else: 

206 typeName = getFullyQualifiedName(value) 

207 ex = UnsupportedValueTypeException(f"Unsupported type '{typeName}' for key '{key}' in node '{self._key}'.") 

208 ex.add_note(f"The YAML parser returned a value that is neither a scalar (str, int, float), nor a map or sequence.") 

209 raise ex 

210 

211 self._cache[key] = value 

212 

213 return value 

214 

215 def _ResolveVariables(self, value: str) -> str: 

216 """ 

217 Resolve the ``${...}`` variables inside a value. 

218 

219 A variable references another node by a path expression, so a value can be composed from other values of the 

220 same configuration. 

221 

222 :param value: The raw value, possibly containing variables. 

223 :returns: The value with every variable replaced by what it references. 

224 :raises InterpolationException: If a variable is malformed - a dangling ``$`` at the end of the value, or a 

225 missing closing ``}`` for a ``${`` at some position. |br| 

226 Use ``$$`` to escape a literal dollar sign. 

227 :raises KeyNotFoundException: If a referenced key doesn't exist. 

228 """ 

229 if value == "": 229 ↛ 230line 229 didn't jump to line 230 because the condition on line 229 was never true

230 return "" 

231 elif "$" not in value: 

232 return value 

233 

234 rawValue = value 

235 result = "" 

236 

237 while (len(rawValue) > 0): 

238# print(f"_ResolveVariables: LOOP rawValue='{rawValue}'") 

239 beginPos = rawValue.find("$") 

240 if beginPos < 0: 

241 result += rawValue 

242 rawValue = "" 

243 else: 

244 result += rawValue[:beginPos] 

245 if beginPos + 1 >= len(rawValue): 

246 ex = InterpolationException(f"Dangling '$' at the end of value '{value}'.") 

247 ex.add_note(f"Use '$$' to escape a literal dollar sign.") 

248 raise ex 

249 elif rawValue[beginPos + 1] == "$": 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 result += "$" 

251 rawValue = rawValue[1:] 

252 elif rawValue[beginPos + 1] == "{": 252 ↛ 237line 252 didn't jump to line 237 because the condition on line 252 was always true

253 endPos = rawValue.find("}", beginPos) 

254 nextPos = rawValue.rfind("$", beginPos, endPos) 

255 if endPos < 0: 

256 ex = InterpolationException(f"Unclosed variable reference in value '{value}'.") 

257 ex.add_note(f"Missing closing '}}' for the '${{' at position {beginPos}.") 

258 raise ex 

259 if (nextPos > 0) and (nextPos < endPos): # an embedded $-sign 

260 path = rawValue[nextPos+2:endPos] 

261# print(f"_ResolveVariables: path='{path}'") 

262 innervalue = self._GetValueByPathExpression(self._ToPath(path)) 

263# print(f"_ResolveVariables: innervalue='{innervalue}'") 

264 rawValue = rawValue[beginPos:nextPos] + str(innervalue) + rawValue[endPos + 1:] 

265# print(f"_ResolveVariables: new rawValue='{rawValue}'") 

266 else: 

267 path = rawValue[beginPos+2:endPos] 

268 rawValue = rawValue[endPos+1:] 

269 result += str(self._GetValueByPathExpression(self._ToPath(path))) 

270 

271 return result 

272 

273 def _GetValueByPathExpression(self, path: list[KeyT]) -> ValueT: 

274 """ 

275 Return the value the given path refers to. 

276 

277 :param path: Path elements, where ``..`` selects the parent node. 

278 :returns: The scalar value at that path. 

279 :raises KeyNotFoundException: If a path element doesn't exist. 

280 :raises PathExpressionException: If the path resolves to a node instead of a value. Extend the path expression 

281 to address a scalar value. 

282 """ 

283 node = self 

284 for p in path: 

285 if p == "..": 

286 node = node._parent 

287 else: 

288 node = node._GetNodeOrValue(p) 

289 

290 if isinstance(node, Dictionary): 

291 pathExpression = ":".join(str(element) for element in path) 

292 ex = PathExpressionException(f"Path expression '{pathExpression}' resolves to a dictionary, not to a value.") 

293 ex.add_note(f"Element '{p}' is a dictionary. Extend the path expression to address a scalar value.") 

294 raise ex 

295 

296 return node 

297 

298 def _GetNodeOrValueByPathExpression(self, path: list[KeyT]) -> ValueT: 

299 """ 

300 Return the node or value the given path refers to. 

301 

302 :param path: Path elements, where ``..`` selects the parent node. 

303 :returns: A node or a scalar value at that path. 

304 :raises KeyNotFoundException: If a path element doesn't exist. 

305 """ 

306 node = self 

307 for p in path: 

308 if p == "..": 308 ↛ 309line 308 didn't jump to line 309 because the condition on line 308 was never true

309 node = node._parent 

310 else: 

311 node = node._GetNodeOrValue(p) 

312 

313 return node 

314 

315 

316@export 

317class Dictionary(Node, Abstract_Dict): 

318 """A dictionary node in a YAML data file.""" 

319 

320 _keys: list[KeyT] #: List of keys in this dictionary. 

321 

322 def __init__( 

323 self, 

324 root: Configuration, 

325 parent: NodeT, 

326 key: KeyT, 

327 yamlNode: CommentedMap 

328 ) -> None: 

329 """ 

330 Initializes a YAML dictionary. 

331 

332 :param root: Reference to the root node. 

333 :param parent: Reference to the parent node. 

334 :param key: Key of the node within its parent. 

335 :param yamlNode: Reference to the YAML node. 

336 """ 

337 Node.__init__(self, root, parent, key, yamlNode) 

338 

339 self._keys = [str(k) for k in yamlNode.keys()] 

340 

341 def __contains__(self, key: KeyT) -> bool: 

342 """ 

343 Checks if the key is in this dictionary. 

344 

345 :param key: The key to check. 

346 :returns: ``True``, if the key is in the dictionary. 

347 """ 

348 return key in self._keys 

349 

350 def __iter__(self) -> typing_Iterator[ValueT]: 

351 """ 

352 Returns an iterator to iterate dictionary keys. 

353 

354 :returns: Dictionary key iterator. 

355 """ 

356 

357 class Iterator(metaclass=ExtendedType, slots=True): 

358 """Iterator to iterate dictionary items.""" 

359 

360 _iter: typing_Iterator[ValueT] #: Iterator over the underlying dictionary's keys. 

361 _obj: Dictionary #: The dictionary being iterated. 

362 

363 def __init__(self, obj: Dictionary) -> None: 

364 """ 

365 Initializes an iterator for a YAML dictionary node. 

366 

367 :param obj: YAML dictionary to iterate. 

368 """ 

369 self._iter = iter(obj._keys) 

370 self._obj = obj 

371 

372 def __iter__(self) -> Self: 

373 """ 

374 Return itself to fulfil the iterator protocol. 

375 

376 :returns: Itself. 

377 """ 

378 return self # pragma: no cover 

379 

380 def __next__(self) -> ValueT: 

381 """ 

382 Returns the next item in the dictionary. 

383 

384 :returns: Next item. 

385 """ 

386 key = next(self._iter) 

387 return self._obj[key] 

388 

389 return Iterator(self) 

390 

391 

392@export 

393class Sequence(Node, Abstract_Seq): 

394 """A sequence node (ordered list) in a YAML data file.""" 

395 

396 def __init__( 

397 self, 

398 root: Configuration, 

399 parent: NodeT, 

400 key: KeyT, 

401 yamlNode: CommentedSeq 

402 ) -> None: 

403 """ 

404 Initializes a YAML sequence (list). 

405 

406 :param root: Reference to the root node. 

407 :param parent: Reference to the parent node. 

408 :param key: Key of the node within its parent. 

409 :param yamlNode: Reference to the YAML node. 

410 """ 

411 Node.__init__(self, root, parent, key, yamlNode) 

412 

413 self._length = len(yamlNode) 

414 

415 def __iter__(self) -> typing_Iterator[ValueT]: 

416 """ 

417 Returns an iterator to iterate items in the sequence of sub-nodes. 

418 

419 :returns: Iterator to iterate items in a sequence. 

420 """ 

421 class Iterator(metaclass=ExtendedType, slots=True): 

422 """Iterator to iterate sequence items.""" 

423 

424 _i: int #: internal iterator position 

425 _obj: Sequence #: Sequence object to iterate 

426 

427 def __init__(self, obj: Sequence) -> None: 

428 """ 

429 Initializes an iterator for a YAML sequence node. 

430 

431 :param obj: YAML sequence to iterate. 

432 """ 

433 self._i = 0 

434 self._obj = obj 

435 

436 def __iter__(self) -> Self: 

437 """ 

438 Return itself to fulfil the iterator protocol. 

439 

440 :returns: Itself. 

441 """ 

442 return self # pragma: no cover 

443 

444 def __next__(self) -> ValueT: 

445 """ 

446 Returns the next item in the sequence. 

447 

448 :returns: Next item. 

449 :raises StopIteration: If end of sequence is reached. 

450 """ 

451 if self._i >= len(self._obj): 

452 raise StopIteration 

453 

454 result = self._obj[str(self._i)] 

455 self._i += 1 

456 return result 

457 

458 return Iterator(self) 

459 

460 

461setattr(Node, "DICT_TYPE", Dictionary) 

462setattr(Node, "SEQ_TYPE", Sequence) 

463 

464 

465@export 

466class Configuration(Dictionary, Abstract_Configuration): 

467 """A configuration read from a YAML file.""" 

468 

469 _yamlConfig: YAML #: The parsed YAML document this configuration is based on. 

470 

471 def __init__(self, configFile: Path) -> None: 

472 """ 

473 Initializes a configuration instance that reads a YAML file as input. 

474 

475 All sequence items or dictionaries key-value-pairs in the YAML file are accessible via Python's dictionary syntax. 

476 

477 :param configFile: Configuration file to read and parse. 

478 :raises ConfigurationException: If the YAML file doesn't exist or can't be parsed. 

479 """ 

480 if not configFile.exists(): 480 ↛ 481line 480 didn't jump to line 481 because the condition on line 480 was never true

481 raise ConfigurationException(f"JSON configuration file '{configFile}' not found.") from FileNotFoundError(configFile) 

482 

483 with configFile.open("r", encoding="utf-8") as file: 

484 self._yamlConfig = YAML().load(file) 

485 

486 Dictionary.__init__(self, self, self, None, self._yamlConfig) 

487 Abstract_Configuration.__init__(self, configFile) 

488 

489 def __getitem__(self, key: str) -> ValueT: 

490 """ 

491 Access a configuration node by key. 

492 

493 :param key: The key to look for. 

494 :returns: A node (sequence or dictionary) or scalar value (int, float, str). 

495 """ 

496 return self._GetNodeOrValue(str(key)) 

497 

498 # 

499 # :param key: Key of the value to write. 

500 # :param value: The new value. 

501 # :raises NotImplementedError: Writing a configuration is not supported by this implementation. 

502 # """ 

503 # raise NotImplementedError()