Coverage for pyTooling / Configuration / YAML.py: 90%
165 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-28 12:48 +0000
1# ==================================================================================================================== #
2# _____ _ _ ____ __ _ _ _ #
3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ / ___|___ _ __ / _(_) __ _ _ _ _ __ __ _| |_(_) ___ _ __ #
4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` || | / _ \| '_ \| |_| |/ _` | | | | '__/ _` | __| |/ _ \| '_ \ #
5# | |_) | |_| || | (_) | (_) | | | | | | (_| || |__| (_) | | | | _| | (_| | |_| | | | (_| | |_| | (_) | | | | #
6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)____\___/|_| |_|_| |_|\__, |\__,_|_| \__,_|\__|_|\___/|_| |_| #
7# |_| |___/ |___/ |___/ #
8# ==================================================================================================================== #
9# Authors: #
10# Patrick Lehmann #
11# #
12# License: #
13# ==================================================================================================================== #
14# Copyright 2021-2025 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.
34.. hint::
36 See :ref:`high-level help <CONFIG/FileFormat/YAML>` for explanations and usage examples.
37"""
38from pathlib import Path
39from typing import Dict, List, Union, Iterator as typing_Iterator, Self
41try:
42 from ruamel.yaml import YAML, CommentedMap, CommentedSeq
43except ImportError as ex: # pragma: no cover
44 raise Exception("Optional dependency 'ruamel.yaml' not installed. Either install pyTooling with extra dependencies 'pyTooling[yaml]' or install 'ruamel.yaml' directly.") from ex
46try:
47 from pyTooling.Decorators import export
48 from pyTooling.MetaClasses import ExtendedType
49 from pyTooling.Configuration import ConfigurationException, KeyT, NodeT, ValueT
50 from pyTooling.Configuration import Node as Abstract_Node
51 from pyTooling.Configuration import Dictionary as Abstract_Dict
52 from pyTooling.Configuration import Sequence as Abstract_Seq
53 from pyTooling.Configuration import Configuration as Abstract_Configuration
54except (ImportError, ModuleNotFoundError): # pragma: no cover
55 print("[pyTooling.Configuration.YAML] Could not import from 'pyTooling.*'!")
57 try:
58 from Decorators import export
59 from MetaClasses import ExtendedType
60 from pyTooling.Configuration import ConfigurationException, KeyT, NodeT, ValueT
61 from pyTooling.Configuration import Node as Abstract_Node
62 from pyTooling.Configuration import Dictionary as Abstract_Dict
63 from pyTooling.Configuration import Sequence as Abstract_Seq
64 from pyTooling.Configuration import Configuration as Abstract_Configuration
65 except (ImportError, ModuleNotFoundError) as ex: # pragma: no cover
66 print("[pyTooling.Configuration.YAML] Could not import directly!")
67 raise ex
70@export
71class Node(Abstract_Node):
72 """
73 Node in a YAML configuration data structure.
74 """
76 _yamlNode: Union[CommentedMap, CommentedSeq] #: Reference to the associated YAML node.
77 _cache: Dict[str, ValueT]
78 _key: KeyT #: Key of this node.
79 _length: int #: Number of sub-elements.
81 def __init__(
82 self,
83 root: "Configuration",
84 parent: NodeT,
85 key: KeyT,
86 yamlNode: Union[CommentedMap, CommentedSeq]
87 ) -> None:
88 """
89 Initializes a YAML node.
91 :param root: Reference to the root node.
92 :param parent: Reference to the parent node.
93 :param key:
94 :param yamlNode: Reference to the YAML node.
95 """
96 Abstract_Node.__init__(self, root, parent)
98 self._yamlNode = yamlNode
99 self._cache = {}
100 self._key = key
101 self._length = len(yamlNode)
103 def __len__(self) -> int:
104 """
105 Returns the number of sub-elements.
107 :returns: Number of sub-elements.
108 """
109 return self._length
111 def __getitem__(self, key: KeyT) -> ValueT:
112 """
113 Access an element in the node by index or key.
115 :param key: Index or key of the element.
116 :returns: A node (sequence or dictionary) or scalar value (int, float, str).
117 """
118 return self._GetNodeOrValue(str(key))
120 @property
121 def Key(self) -> KeyT:
122 """
123 Property to access the node's key.
125 :returns: Key of the node.
126 """
127 return self._key
129 @Key.setter
130 def Key(self, value: KeyT) -> None:
131 raise NotImplementedError()
133 def QueryPath(self, query: str) -> ValueT:
134 """
135 Return a node or value based on a path description to that node or value.
137 :param query: String describing the path to the node or value.
138 :returns: A node (sequence or dictionary) or scalar value (int, float, str).
139 """
140 path = self._ToPath(query)
141 return self._GetNodeOrValueByPathExpression(path)
143 @staticmethod
144 def _ToPath(query: str) -> List[Union[str, int]]:
145 return query.split(":")
147 def _GetNodeOrValue(self, key: str) -> ValueT:
148 try:
149 value = self._cache[key]
150 except KeyError:
151 try:
152 value = self._yamlNode[key]
153 except (KeyError, TypeError):
154 try:
155 value = self._yamlNode[int(key)]
156 except KeyError:
157 try:
158 value = self._yamlNode[float(key)]
159 except KeyError as ex:
160 raise Exception(f"") from ex # XXX: needs error message
162 if isinstance(value, str):
163 value = self._ResolveVariables(value)
164 elif isinstance(value, (int, float)):
165 value = str(value)
166 elif isinstance(value, CommentedMap):
167 value = self.DICT_TYPE(self, self, key, value)
168 elif isinstance(value, CommentedSeq): 168 ↛ 171line 168 didn't jump to line 171 because the condition on line 168 was always true
169 value = self.SEQ_TYPE(self, self, key, value)
170 else:
171 raise Exception(f"") from TypeError(f"Unknown type '{value.__class__.__name__}' returned from ruamel.yaml.") # XXX: error message
173 self._cache[key] = value
175 return value
177 def _ResolveVariables(self, value: str) -> str:
178 if value == "": 178 ↛ 179line 178 didn't jump to line 179 because the condition on line 178 was never true
179 return ""
180 elif "$" not in value:
181 return value
183 rawValue = value
184 result = ""
186 while (len(rawValue) > 0):
187# print(f"_ResolveVariables: LOOP rawValue='{rawValue}'")
188 beginPos = rawValue.find("$")
189 if beginPos < 0:
190 result += rawValue
191 rawValue = ""
192 else:
193 result += rawValue[:beginPos]
194 if rawValue[beginPos + 1] == "$": 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true
195 result += "$"
196 rawValue = rawValue[1:]
197 elif rawValue[beginPos + 1] == "{": 197 ↛ 186line 197 didn't jump to line 186 because the condition on line 197 was always true
198 endPos = rawValue.find("}", beginPos)
199 nextPos = rawValue.rfind("$", beginPos, endPos)
200 if endPos < 0: 200 ↛ 201line 200 didn't jump to line 201 because the condition on line 200 was never true
201 raise Exception(f"") # XXX: InterpolationSyntaxError(option, section, f"Bad interpolation variable reference {rest!r}")
202 if (nextPos > 0) and (nextPos < endPos): # an embedded $-sign
203 path = rawValue[nextPos+2:endPos]
204# print(f"_ResolveVariables: path='{path}'")
205 innervalue = self._GetValueByPathExpression(self._ToPath(path))
206# print(f"_ResolveVariables: innervalue='{innervalue}'")
207 rawValue = rawValue[beginPos:nextPos] + str(innervalue) + rawValue[endPos + 1:]
208# print(f"_ResolveVariables: new rawValue='{rawValue}'")
209 else:
210 path = rawValue[beginPos+2:endPos]
211 rawValue = rawValue[endPos+1:]
212 result += str(self._GetValueByPathExpression(self._ToPath(path)))
214 return result
216 def _GetValueByPathExpression(self, path: List[KeyT]) -> ValueT:
217 node = self
218 for p in path:
219 if p == "..":
220 node = node._parent
221 else:
222 node = node._GetNodeOrValue(p)
224 if isinstance(node, Dictionary): 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true
225 raise Exception(f"Error when resolving path expression '{':'.join(path)}' at '{p}'.") from TypeError(f"") # XXX: needs error messages
227 return node
229 def _GetNodeOrValueByPathExpression(self, path: List[KeyT]) -> ValueT:
230 node = self
231 for p in path:
232 if p == "..": 232 ↛ 233line 232 didn't jump to line 233 because the condition on line 232 was never true
233 node = node._parent
234 else:
235 node = node._GetNodeOrValue(p)
237 return node
240@export
241class Dictionary(Node, Abstract_Dict):
242 """A dictionary node in a YAML data file."""
244 _keys: List[KeyT] #: List of keys in this dictionary.
246 def __init__(
247 self,
248 root: "Configuration",
249 parent: NodeT,
250 key: KeyT,
251 yamlNode: CommentedMap
252 ) -> None:
253 """
254 Initializes a YAML dictionary.
256 :param root: Reference to the root node.
257 :param parent: Reference to the parent node.
258 :param key:
259 :param yamlNode: Reference to the YAML node.
260 """
261 Node.__init__(self, root, parent, key, yamlNode)
263 self._keys = [str(k) for k in yamlNode.keys()]
265 def __contains__(self, key: KeyT) -> bool:
266 """
267 Checks if the key is in this dictionary.
269 :param key: The key to check.
270 :returns: ``True``, if the key is in the dictionary.
271 """
272 return key in self._keys
274 def __iter__(self) -> typing_Iterator[ValueT]:
275 """
276 Returns an iterator to iterate dictionary keys.
278 :returns: Dictionary key iterator.
279 """
281 class Iterator(metaclass=ExtendedType, slots=True):
282 """Iterator to iterate dictionary items."""
284 _iter: typing_Iterator[ValueT]
285 _obj: Dictionary
287 def __init__(self, obj: Dictionary) -> None:
288 """
289 Initializes an iterator for a YAML dictionary node.
291 :param obj: YAML dictionary to iterate.
292 """
293 self._iter = iter(obj._keys)
294 self._obj = obj
296 def __iter__(self) -> Self:
297 """
298 Return itself to fulfil the iterator protocol.
300 :returns: Itself.
301 """
302 return self # pragma: no cover
304 def __next__(self) -> ValueT:
305 """
306 Returns the next item in the dictionary.
308 :returns: Next item.
309 """
310 key = next(self._iter)
311 return self._obj[key]
313 return Iterator(self)
316@export
317class Sequence(Node, Abstract_Seq):
318 """A sequence node (ordered list) in a YAML data file."""
320 def __init__(
321 self,
322 root: "Configuration",
323 parent: NodeT,
324 key: KeyT,
325 yamlNode: CommentedSeq
326 ) -> None:
327 """
328 Initializes a YAML sequence (list).
330 :param root: Reference to the root node.
331 :param parent: Reference to the parent node.
332 :param key:
333 :param yamlNode: Reference to the YAML node.
334 """
335 Node.__init__(self, root, parent, key, yamlNode)
337 self._length = len(yamlNode)
339 def __iter__(self) -> typing_Iterator[ValueT]:
340 """
341 Returns an iterator to iterate items in the sequence of sub-nodes.
343 :returns: Iterator to iterate items in a sequence.
344 """
345 class Iterator(metaclass=ExtendedType, slots=True):
346 """Iterator to iterate sequence items."""
348 _i: int #: internal iterator position
349 _obj: Sequence #: Sequence object to iterate
351 def __init__(self, obj: Sequence) -> None:
352 """
353 Initializes an iterator for a YAML sequence node.
355 :param obj: YAML sequence to iterate.
356 """
357 self._i = 0
358 self._obj = obj
360 def __iter__(self) -> Self:
361 """
362 Return itself to fulfil the iterator protocol.
364 :returns: Itself.
365 """
366 return self # pragma: no cover
368 def __next__(self) -> ValueT:
369 """
370 Returns the next item in the sequence.
372 :returns: Next item.
373 :raises StopIteration: If end of sequence is reached.
374 """
375 try:
376 result = self._obj[str(self._i)]
377 self._i += 1
378 return result
379 except IndexError:
380 raise StopIteration
382 return Iterator(self)
385setattr(Node, "DICT_TYPE", Dictionary)
386setattr(Node, "SEQ_TYPE", Sequence)
389@export
390class Configuration(Dictionary, Abstract_Configuration):
391 """A configuration read from a YAML file."""
393 _yamlConfig: YAML
395 def __init__(self, configFile: Path) -> None:
396 """
397 Initializes a configuration instance that reads a YAML file as input.
399 All sequence items or dictionaries key-value-pairs in the YAML file are accessible via Python's dictionary syntax.
401 :param configFile: Configuration file to read and parse.
402 """
403 if not configFile.exists(): 403 ↛ 404line 403 didn't jump to line 404 because the condition on line 403 was never true
404 raise ConfigurationException(f"JSON configuration file '{configFile}' not found.") from FileNotFoundError(configFile)
406 with configFile.open("r", encoding="utf-8") as file:
407 self._yamlConfig = YAML().load(file)
409 Dictionary.__init__(self, self, self, None, self._yamlConfig)
410 Abstract_Configuration.__init__(self, configFile)
412 def __getitem__(self, key: str) -> ValueT:
413 """
414 Access a configuration node by key.
416 :param key: The key to look for.
417 :returns: A node (sequence or dictionary) or scalar value (int, float, str).
418 """
419 return self._GetNodeOrValue(str(key))
421 def __setitem__(self, key: str, value: ValueT) -> None:
422 raise NotImplementedError()