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

167 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-04-25 22:22 +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. 

33 

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

35""" 

36from pathlib import Path 

37from typing import Dict, List, Union, Iterator as typing_Iterator 

38 

39try: 

40 from ruamel.yaml import YAML, CommentedMap, CommentedSeq 

41except ImportError as ex: # pragma: no cover 

42 raise Exception(f"Optional dependency 'ruamel.yaml' not installed. Either install pyTooling with extra dependencies 'pyTooling[yaml]' or install 'ruamel.yaml' directly.") from ex 

43 

44try: 

45 from pyTooling.Decorators import export 

46 from pyTooling.MetaClasses import ExtendedType 

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

48 from pyTooling.Configuration import Node as Abstract_Node 

49 from pyTooling.Configuration import Dictionary as Abstract_Dict 

50 from pyTooling.Configuration import Sequence as Abstract_Seq 

51 from pyTooling.Configuration import Configuration as Abstract_Configuration 

52except (ImportError, ModuleNotFoundError): # pragma: no cover 

53 print("[pyTooling.Configuration.YAML] Could not import from 'pyTooling.*'!") 

54 

55 try: 

56 from Decorators import export 

57 from MetaClasses import ExtendedType 

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

59 from pyTooling.Configuration import Node as Abstract_Node 

60 from pyTooling.Configuration import Dictionary as Abstract_Dict 

61 from pyTooling.Configuration import Sequence as Abstract_Seq 

62 from pyTooling.Configuration import Configuration as Abstract_Configuration 

63 except (ImportError, ModuleNotFoundError) as ex: # pragma: no cover 

64 print("[pyTooling.Configuration.YAML] Could not import directly!") 

65 raise ex 

66 

67 

68@export 

69class Node(Abstract_Node): 

70 _yamlNode: Union[CommentedMap, CommentedSeq] 

71 _cache: Dict[str, ValueT] 

72 _key: KeyT 

73 _length: int 

74 

75 def __init__(self, root: "Configuration", parent: NodeT, key: KeyT, yamlNode: Union[CommentedMap, CommentedSeq]) -> None: 

76 Abstract_Node.__init__(self, root, parent) 

77 

78 self._yamlNode = yamlNode 

79 self._cache = {} 

80 self._key = key 

81 self._length = len(yamlNode) 

82 

83 def __len__(self) -> int: 

84 """ 

85 Returns the number of sub-elements. 

86 

87 :returns: Number of sub-elements. 

88 """ 

89 return self._length 

90 

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

92 return self._GetNodeOrValue(str(key)) 

93 

94 @property 

95 def Key(self) -> KeyT: 

96 return self._key 

97 

98 @Key.setter 

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

100 raise NotImplementedError() 

101 

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

103 path = self._ToPath(query) 

104 return self._GetNodeOrValueByPathExpression(path) 

105 

106 @staticmethod 

107 def _ToPath(query: str) -> List[Union[str, int]]: 

108 return query.split(":") 

109 

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

111 try: 

112 value = self._cache[key] 

113 except KeyError: 

114 try: 

115 value = self._yamlNode[key] 

116 except (KeyError, TypeError): 

117 try: 

118 value = self._yamlNode[int(key)] 

119 except KeyError: 

120 try: 

121 value = self._yamlNode[float(key)] 

122 except KeyError as ex: 

123 raise Exception(f"") from ex # XXX: needs error message 

124 

125 if isinstance(value, str): 

126 value = self._ResolveVariables(value) 

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

128 value = str(value) 

129 elif isinstance(value, CommentedMap): 

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

131 elif isinstance(value, CommentedSeq): 131 ↛ 134line 131 didn't jump to line 134 because the condition on line 131 was always true

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

133 else: 

134 raise Exception(f"") from TypeError(f"Unknown type '{value.__class__.__name__}' returned from ruamel.yaml.") # XXX: error message 

135 

136 self._cache[key] = value 

137 

138 return value 

139 

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

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

142 return "" 

143 elif "$" not in value: 

144 return value 

145 

146 rawValue = value 

147 result = "" 

148 

149 while (len(rawValue) > 0): 

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

151 beginPos = rawValue.find("$") 

152 if beginPos < 0: 

153 result += rawValue 

154 rawValue = "" 

155 else: 

156 result += rawValue[:beginPos] 

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

158 result += "$" 

159 rawValue = rawValue[1:] 

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

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

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

163 if endPos < 0: 163 ↛ 164line 163 didn't jump to line 164 because the condition on line 163 was never true

164 raise Exception(f"") # XXX: InterpolationSyntaxError(option, section, f"Bad interpolation variable reference {rest!r}") 

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

166 path = rawValue[nextPos+2:endPos] 

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

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

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

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

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

172 else: 

173 path = rawValue[beginPos+2:endPos] 

174 rawValue = rawValue[endPos+1:] 

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

176 

177 return result 

178 

179 def _GetValueByPathExpression(self, path: List[KeyT]) -> ValueT: 

180 node = self 

181 for p in path: 

182 if p == "..": 

183 node = node._parent 

184 else: 

185 node = node._GetNodeOrValue(p) 

186 

187 if isinstance(node, Dictionary): 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true

188 raise Exception(f"Error when resolving path expression '{':'.join(path)}' at '{p}'.") from TypeError(f"") # XXX: needs error messages 

189 

190 return node 

191 

192 def _GetNodeOrValueByPathExpression(self, path: List[KeyT]) -> ValueT: 

193 node = self 

194 for p in path: 

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

196 node = node._parent 

197 else: 

198 node = node._GetNodeOrValue(p) 

199 

200 return node 

201 

202 

203@export 

204class Dictionary(Node, Abstract_Dict): 

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

206 

207 _keys: List[KeyT] 

208 

209 def __init__(self, root: "Configuration", parent: NodeT, key: KeyT, yamlNode: CommentedMap) -> None: 

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

211 

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

213 

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

215 return key in self._keys 

216 

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

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

219 _iter: typing_Iterator[ValueT] 

220 _obj: Dictionary 

221 

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

223 self._iter = iter(obj._keys) 

224 self._obj = obj 

225 

226 def __iter__(self) -> "Iterator": 

227 """ 

228 Return itself to fulfil the iterator protocol. 

229 

230 :returns: Itself. 

231 """ 

232 return self # pragma: no cover 

233 

234 def __next__(self) -> ValueT: 

235 """ 

236 Returns the next item in the dictionary. 

237 

238 :returns: Next item. 

239 """ 

240 key = next(self._iter) 

241 return self._obj[key] 

242 

243 return Iterator(self) 

244 

245 

246@export 

247class Sequence(Node, Abstract_Seq): 

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

249 

250 def __init__(self, root: "Configuration", parent: NodeT, key: KeyT, yamlNode: CommentedSeq) -> None: 

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

252 

253 self._length = len(yamlNode) 

254 

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

256 return self._GetNodeOrValue(str(key)) 

257 

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

259 """ 

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

261 

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

263 """ 

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

265 """Iterator to iterate sequence items.""" 

266 

267 _i: int #: internal iterator position 

268 _obj: Sequence #: Sequence object to iterate 

269 

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

271 self._i = 0 

272 self._obj = obj 

273 

274 def __iter__(self) -> "Iterator": 

275 """ 

276 Return itself to fulfil the iterator protocol. 

277 

278 :returns: Itself. 

279 """ 

280 return self # pragma: no cover 

281 

282 def __next__(self) -> ValueT: 

283 """ 

284 Returns the next item in the sequence. 

285 

286 :returns: Next item. 

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

288 """ 

289 try: 

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

291 self._i += 1 

292 return result 

293 except IndexError: 

294 raise StopIteration 

295 

296 return Iterator(self) 

297 

298 

299setattr(Node, "DICT_TYPE", Dictionary) 

300setattr(Node, "SEQ_TYPE", Sequence) 

301 

302 

303@export 

304class Configuration(Dictionary, Abstract_Configuration): 

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

306 

307 _yamlConfig: YAML 

308 

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

310 """ 

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

312 

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

314 

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

316 """ 

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

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

319 

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

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

322 

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

324 Abstract_Configuration.__init__(self, configFile) 

325 

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

327 """ 

328 Access a configuration node by key. 

329 

330 :param key: The key to look for. 

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

332 """ 

333 return self._GetNodeOrValue(str(key)) 

334 

335 def __setitem__(self, key: str, value: ValueT) -> None: 

336 raise NotImplementedError()