Coverage for yamlable/base.py: 75%

1# Authors: Sylvain MARIE <sylvain.marie@se.com> 

2# + All contributors to <https://github.com/smarie/python-yamlable> 

3# 

4# License: 3-clause BSD, <https://github.com/smarie/python-yamlable/blob/master/LICENSE> 

5 

6from abc import ABCMeta 

7from collections import OrderedDict 

8 

9from yaml import ScalarNode, SequenceNode, MappingNode 

10 

11try: 

12 # Python 2 only: 

13 from StringIO import StringIO as _StringIO # type: ignore # noqa 

14 

15 # create a variant that can serve as a context manager 

16 class StringIO(_StringIO): 

17 def __enter__(self): 

18 return self 

19 

20 def __exit__(self, exception_type, exception_value, traceback): 

21 self.close() 

22 

23except ImportError: 

24 # (IOBase is only used in type hints) 

25 from io import IOBase, StringIO # type: ignore 

26 

27from warnings import warn 

28 

29import six 

30 

31try: # python 3.5+ 

32 from typing import Union, TypeVar, Dict, Any, Sequence 

33 

34 Y = TypeVar('Y', bound='AbstractYamlObject') 

35 

36except ImportError: 

37 pass 

38 

39try: # python 3.5.4+ 

40 from typing import Type 

41except ImportError: 

42 pass # normal for old versions of typing 

43 

44 

45class AbstractYamlObject(six.with_metaclass(ABCMeta, object)): 

46 """ 

47 Adds convenient methods load(s)_yaml/dump(s)_yaml to any object, to call pyyaml features directly on the object or 

48 on the object class. 

49 

50 Also adds the two methods __to_yaml_dict__ / __from_yaml_dict__, that are common to YamlObject2 and YamlAble. 

51 Default implementation uses vars(self) and cls(**dct), but subclasses can override. 

52 """ 

53 

54 # def __to_yaml_scalar__(self): 

55 # # type: (...) -> Any 

56 # """ 

57 # Implementors should transform the object into a scalar containing all information necessary to decode the 

58 # object as a YAML scalar in the future. 

59 # 

60 # Default implementation raises an error. 

61 # :return: 

62 # """ 

63 # raise NotImplementedError("Please override `__to_yaml_scalar__` if you wish to dump instances of `%s`" 

64 # " as yaml scalars." % type(self).__name__) 

65 # 

66 # def __to_yaml_list__(self): 

67 # # type: (...) -> Sequence[Any] 

68 # """ 

69 # Implementors should transform the object into a Sequence containing all information necessary to decode the 

70 # object as a YAML sequence in the future. 

71 # 

72 # Default implementation raises an error. 

73 # :return: 

74 # """ 

75 # raise NotImplementedError("Please override `__to_yaml_list__` if you wish to dump instances of `%s`" 

76 # " as yaml sequences." % type(self).__name__) 

77 

78 def __to_yaml_dict__(self): 

79 # type: (...) -> Dict[str, Any] 

80 """ 

81 Implementors should transform the object into a dictionary containing all information necessary to decode the 

82 object in the future. That dictionary will be serialized as a YAML mapping. 

83 

84 Default implementation returns vars(self). 

85 :return: 

86 """ 

87 # Legacy compliance with old 'not dunder' method name TODO remove in future version 

88 if 'to_yaml_dict' in dir(self): 

89 warn(type(self).__name__ + " still uses the legacy method name 'to_yaml_dict'. This name will not be " 

90 "supported in future version, please use '__to_yaml_dict__' instead") 

91 return self.to_yaml_dict() # type: ignore 

92 

93 # Default: return vars(self) (Note: no need to make a copy, pyyaml does not modify it) 

94 return vars(self) 

95 

96 @classmethod 

97 def __from_yaml_scalar__(cls, # type: Type[Y] 

98 scalar, # type: Any 

99 yaml_tag # type: str 

100 ): 

101 # type: (...) -> Y 

102 """ 

103 Implementors should transform the given scalar (read from yaml by the pyYaml stack) into an object instance. 

104 The yaml tag associated to this object, read in the yaml document, is provided in parameter. 

105 

106 Note that for YamlAble and YamlObject2 subclasses, if this method is called the yaml tag will already have 

107 been checked so implementors do not have to validate it. 

108 

109 Default implementation returns cls(scalar) 

110 

111 :param scalar: the yaml scalar 

112 :param yaml_tag: the yaml schema id that was used for encoding the object (it has already been checked 

113 against is_json_schema_id_supported) 

114 :return: 

115 """ 

116 # Default: call constructor with positional arguments 

117 return cls(scalar) # type: ignore 

118 

119 @classmethod 

120 def __from_yaml_list__(cls, # type: Type[Y] 

121 seq, # type: Sequence[Any] 

122 yaml_tag # type: str 

123 ): 

124 # type: (...) -> Y 

125 """ 

126 Implementors should transform the given Sequence (read from yaml by the pyYaml stack) into an object instance. 

127 The yaml tag associated to this object, read in the yaml document, is provided in parameter. 

128 

129 Note that for YamlAble and YamlObject2 subclasses, if this method is called the yaml tag will already have 

130 been checked so implementors do not have to validate it. 

131 

132 Default implementation returns cls(*seq) 

133 

134 :param seq: the yaml sequence 

135 :param yaml_tag: the yaml schema id that was used for encoding the object (it has already been checked 

136 against is_json_schema_id_supported) 

137 :return: 

138 """ 

139 # Default: call constructor with positional arguments 

140 return cls(*seq) # type: ignore 

141 

142 @classmethod 

143 def __from_yaml_dict__(cls, # type: Type[Y] 

144 dct, # type: Dict[str, Any] 

145 yaml_tag # type: str 

146 ): 

147 # type: (...) -> Y 

148 """ 

149 Implementors should transform the given dictionary (read from yaml by the pyYaml stack) into an object instance. 

150 The yaml tag associated to this object, read in the yaml document, is provided in parameter. 

151 

152 Note that for YamlAble and YamlObject2 subclasses, if this method is called the yaml tag will already have 

153 been checked so implementors do not have to validate it. 

154 

155 Default implementation returns cls(**dct) 

156 

157 :param dct: 

158 :param yaml_tag: the yaml schema id that was used for encoding the object (it has already been checked 

159 against is_json_schema_id_supported) 

160 :return: 

161 """ 

162 # Legacy compliance with old 'not dunder' method name TODO remove in future version 

163 if 'from_yaml_dict' in dir(cls): 

164 warn(cls.__name__ + " still uses the legacy method name 'from_yaml_dict'. This name will not be " 

165 "supported in future version, please use '__from_yaml_dict__' instead") 

166 return cls.from_yaml_dict(dct, yaml_tag) # type: ignore 

167 

168 # Default: call constructor with all keyword arguments 

169 return cls(**dct) # type: ignore 

170 

171 def dump_yaml(self, 

172 file_path_or_stream, # type: Union[str, IOBase, StringIO] 

173 safe=True, # type: bool 

174 **pyyaml_kwargs # type: Any 

175 ): 

176 # type: (...) -> None 

177 """ 

178 Dumps this object to a yaml file or stream using pyYaml. 

179 

180 :param file_path_or_stream: either a string representing the file path, or a stream where to write 

181 :param safe: True (default) uses `yaml.safe_dump`. False uses `yaml.dump` 

182 :param pyyaml_kwargs: keyword arguments for the pyYaml dump method 

183 :return: 

184 """ 

185 from yaml import safe_dump, dump 

186 if isinstance(file_path_or_stream, str): 186 ↛ 187line 186 didn't jump to line 187, because the condition on line 186 was never true

187 with open(file_path_or_stream, mode='w+t') as f: 

188 if safe: 

189 safe_dump(self, f, **pyyaml_kwargs) 

190 else: 

191 dump(self, f, **pyyaml_kwargs) 

192 else: 

193 with file_path_or_stream as f: # type: ignore 

194 if safe: 194 ↛ 197line 194 didn't jump to line 197, because the condition on line 194 was never false

195 safe_dump(self, f, **pyyaml_kwargs) 

196 else: 

197 dump(self, f, **pyyaml_kwargs) 

198 

199 def dumps_yaml(self, 

200 safe=True, # type: bool 

201 **pyyaml_kwargs # type: Any 

202 ): 

203 # type: (...) -> str 

204 """ 

205 Dumps this object to a yaml string and returns it. 

206 

207 :param pyyaml_kwargs: keyword arguments for the pyYaml dump method 

208 :param safe: True (default) uses `yaml.safe_dump`. False uses `yaml.dump` 

209 :return: 

210 """ 

211 from yaml import safe_dump, dump 

212 if safe: 

213 return safe_dump(self, **pyyaml_kwargs) 

214 else: 

215 return dump(self, **pyyaml_kwargs) 

216 

217 @classmethod 

218 def loads_yaml(cls, # type: Type[Y] 

219 yaml_str, # type: str 

220 safe=True # type: bool 

221 ): 

222 # type: (...) -> Y 

223 """ 

224 Utility method to load an instance of this class from the provided yaml string. This methods only returns 

225 successfully if the result is an instance of `cls`. 

226 

227 :param yaml_str: 

228 :param safe: True (default) uses `yaml.safe_load`. False uses `yaml.load` 

229 :return: 

230 """ 

231 return cls.load_yaml(StringIO(yaml_str), safe=safe) 

232 

233 @classmethod 

234 def load_yaml(cls, # type: Type[Y] 

235 file_path_or_stream, # type: Union[str, IOBase, StringIO] 

236 safe=True # type: bool 

237 ): 

238 # type: (...) -> Y 

239 """ 

240 Parses the given file path or stream as a yaml document. This methods only returns successfully if the result 

241 is an instance of `cls`. 

242 

243 :param file_path_or_stream: 

244 :param safe: True (default) uses `yaml.safe_load`. False uses `yaml.load` 

245 :return: 

246 """ 

247 from yaml import safe_load, load 

248 if isinstance(file_path_or_stream, str): 248 ↛ 249line 248 didn't jump to line 249, because the condition on line 248 was never true

249 with open(file_path_or_stream, mode='rt') as f: 

250 if safe: 

251 res = safe_load(f.read()) 

252 else: 

253 res = load(f.read()) 

254 else: 

255 with file_path_or_stream as f: # type: ignore 

256 if safe: 256 ↛ 259line 256 didn't jump to line 259, because the condition on line 256 was never false

257 res = safe_load(f.read()) 

258 else: 

259 res = load(f.read()) 

260 

261 if isinstance(res, cls): 261 ↛ 264line 261 didn't jump to line 264, because the condition on line 261 was never false

262 return res 

263 else: 

264 raise TypeError("Decoded object is not an instance of {}, but a {}. Please make sure that the YAML document" 

265 " starts with the tag defined in you class' `yaml_tag` field, for example `!my_type`" 

266 "".format(cls.__name__, type(res).__name__)) 

267 

268 

269NONE_IGNORE_CHECKS = None 

270# """Tag to be used as yaml tag for abstract classes, to indicate that 

271# they are abstract (checks disabled). Not used anymore, kept for legacy reasons""" 

272 

273 

274def read_yaml_node_as_dict(loader, node): 

275 # type: (...) -> OrderedDict 

276 """ 

277 Utility method to read a yaml node into a dictionary 

278 

279 :param loader: 

280 :param node: 

281 :return: 

282 """ 

283 # loader.flatten_mapping(node) 

284 # pairs = loader.construct_pairs(node, deep=True) # 'deep' allows the construction to be complete (inner seq...) 

285 pairs = loader.construct_mapping(node, deep=True) # 'deep' allows the construction to be complete (inner seq...) 

286 constructor_args = OrderedDict(pairs) 

287 return constructor_args 

288 

289 

290def read_yaml_node_as_sequence(loader, node): 

291 # type: (...) -> Sequence 

292 """ 

293 Utility method to read a yaml node into a sequence 

294 

295 :param loader: 

296 :param node: 

297 :return: 

298 """ 

299 seq = loader.construct_sequence(node, deep=True) # 'deep' allows the construction to be complete (inner seq...) 

300 return seq 

301 

302 

303def read_yaml_node_as_scalar(loader, node): 

304 # type: (...) -> Any 

305 """ 

306 Utility method to read a yaml node into a sequence 

307 

308 :param loader: 

309 :param node: 

310 :return: 

311 """ 

312 value = loader.construct_scalar(node) 

313 return value 

314 

315 

316def read_yaml_node_as_yamlobject( 

317 cls, # type: Type[AbstractYamlObject] 

318 loader, 

319 node, # type: MappingNode 

320 yaml_tag # type: str 

321): 

322 # type: (...) -> AbstractYamlObject 

323 """ 

324 Default implementation: loads the node as a dictionary and calls __from_yaml_dict__ with this dictionary 

325 

326 :param loader: 

327 :param node: 

328 :return: 

329 """ 

330 if isinstance(node, ScalarNode): 

331 constructor_args = read_yaml_node_as_scalar(loader, node) 

332 return cls.__from_yaml_scalar__(constructor_args, yaml_tag=yaml_tag) # type: ignore 

333 

334 elif isinstance(node, SequenceNode): 

335 constructor_args = read_yaml_node_as_sequence(loader, node) 

336 return cls.__from_yaml_list__(constructor_args, yaml_tag=yaml_tag) # type: ignore 

337 

338 elif isinstance(node, MappingNode): 338 ↛ 343line 338 didn't jump to line 343, because the condition on line 338 was never false

339 constructor_args = read_yaml_node_as_dict(loader, node) 

340 return cls.__from_yaml_dict__(constructor_args, yaml_tag=yaml_tag) # type: ignore 

341 

342 else: 

343 raise TypeError("Unknown type of yaml node: %r. Please report this to `yamlable` project." % type(node))