codekingpro/portable-devtools
115k
1 2from __future__ import annotations3 4import datetime5from datetime import timedelta as TimeDelta6import binascii7import sys8import types9import warnings10from collections.abc import Hashable, MutableSequence, MutableMapping11 12# fmt: off13from ruamel.yaml.error import (MarkedYAMLError, MarkedYAMLFutureWarning,14 MantissaNoDotYAML1_1Warning)15from ruamel.yaml.nodes import * # NOQA16from ruamel.yaml.nodes import (SequenceNode, MappingNode, ScalarNode)17from ruamel.yaml.compat import (builtins_module, # NOQA18 nprint, nprintf, version_tnf)19from ruamel.yaml.compat import ordereddict20 21from ruamel.yaml.tag import Tag22from ruamel.yaml.comments import * # NOQA23from ruamel.yaml.comments import (CommentedMap, CommentedOrderedMap, CommentedSet,24 CommentedKeySeq, CommentedSeq, TaggedScalar,25 CommentedKeyMap,26 C_KEY_PRE, C_KEY_EOL, C_KEY_POST,27 C_VALUE_PRE, C_VALUE_EOL, C_VALUE_POST,28 )29from ruamel.yaml.scalarstring import (SingleQuotedScalarString, DoubleQuotedScalarString,30 LiteralScalarString, FoldedScalarString,31 PlainScalarString, ScalarString)32from ruamel.yaml.scalarint import ScalarInt, BinaryInt, OctalInt, HexInt, HexCapsInt33from ruamel.yaml.scalarfloat import ScalarFloat34from ruamel.yaml.scalarbool import ScalarBoolean35from ruamel.yaml.timestamp import TimeStamp36from ruamel.yaml.util import timestamp_regexp, create_timestamp37 38if False: # MYPY39 from typing import Any, Dict, List, Set, Iterator, Union, Optional # NOQA40 41 42__all__ = ['BaseConstructor', 'SafeConstructor', 'Constructor',43 'ConstructorError', 'RoundTripConstructor']44# fmt: on45 46 47class ConstructorError(MarkedYAMLError):48 pass49 50 51class DuplicateKeyFutureWarning(MarkedYAMLFutureWarning):52 pass53 54 55DUPKEY_URL = 'https://yaml.dev/doc/ruamel.yaml/api/#Duplicate_keys'56 57 58class DuplicateKeyError(MarkedYAMLError):59 pass60 61 62class BaseConstructor:63 64 yaml_constructors = {} # type: Dict[Any, Any]65 yaml_multi_constructors = {} # type: Dict[Any, Any]66 67 def __init__(self, preserve_quotes: Optional[bool] = None, loader: Any = None) -> None:68 self.loader = loader69 if self.loader is not None and getattr(self.loader, '_constructor', None) is None:70 self.loader._constructor = self71 self.loader = loader72 self.yaml_base_dict_type = dict73 self.yaml_base_list_type = list74 self.constructed_objects: Dict[Any, Any] = {}75 self.recursive_objects: Dict[Any, Any] = {}76 self.state_generators: List[Any] = []77 self.deep_construct = False78 self._preserve_quotes = preserve_quotes79 self.allow_duplicate_keys = version_tnf((0, 15, 1), (0, 16))80 81 @property82 def composer(self) -> Any:83 if hasattr(self.loader, 'typ'):84 return self.loader.composer85 try:86 return self.loader._composer87 except AttributeError:88 sys.stdout.write(f'slt {type(self)}\n')89 sys.stdout.write(f'slc {self.loader._composer}\n')90 sys.stdout.write(f'{dir(self)}\n')91 raise92 93 @property94 def resolver(self) -> Any:95 if hasattr(self.loader, 'typ'):96 return self.loader.resolver97 return self.loader._resolver98 99 @property100 def scanner(self) -> Any:101 # needed to get to the expanded comments102 if hasattr(self.loader, 'typ'):103 return self.loader.scanner104 return self.loader._scanner105 106 def check_data(self) -> Any:107 # If there are more documents available?108 return self.composer.check_node()109 110 def get_data(self) -> Any:111 # Construct and return the next document.112 if self.composer.check_node():113 return self.construct_document(self.composer.get_node())114 115 def get_single_data(self) -> Any:116 # Ensure that the stream contains a single document and construct it.117 node = self.composer.get_single_node()118 if node is not None:119 return self.construct_document(node)120 return None121 122 def construct_document(self, node: Any) -> Any:123 data = self.construct_object(node)124 while bool(self.state_generators):125 state_generators = self.state_generators126 self.state_generators = []127 for generator in state_generators:128 for _dummy in generator:129 pass130 self.constructed_objects = {}131 self.recursive_objects = {}132 self.deep_construct = False133 return data134 135 def construct_object(self, node: Any, deep: bool = False) -> Any:136 """deep is True when creating an object/mapping recursively,137 in that case want the underlying elements available during construction138 """139 if node in self.constructed_objects:140 return self.constructed_objects[node]141 if deep:142 old_deep = self.deep_construct143 self.deep_construct = True144 if node in self.recursive_objects:145 return self.recursive_objects[node]146 # raise ConstructorError(147 # None, None, 'found unconstructable recursive node', node.start_mark148 # )149 self.recursive_objects[node] = None150 data = self.construct_non_recursive_object(node)151 152 self.constructed_objects[node] = data153 del self.recursive_objects[node]154 if deep:155 self.deep_construct = old_deep156 return data157 158 def construct_non_recursive_object(self, node: Any, tag: Optional[str] = None) -> Any:159 constructor: Any = None160 tag_suffix = None161 if tag is None:162 tag = node.tag163 if tag in self.yaml_constructors:164 constructor = self.yaml_constructors[tag]165 else:166 for tag_prefix in self.yaml_multi_constructors:167 if tag.startswith(tag_prefix):168 tag_suffix = tag[len(tag_prefix) :]169 constructor = self.yaml_multi_constructors[tag_prefix]170 break171 else:172 if None in self.yaml_multi_constructors:173 tag_suffix = tag174 constructor = self.yaml_multi_constructors[None]175 elif None in self.yaml_constructors:176 constructor = self.yaml_constructors[None]177 elif isinstance(node, ScalarNode):178 constructor = self.__class__.construct_scalar179 elif isinstance(node, SequenceNode):180 constructor = self.__class__.construct_sequence181 elif isinstance(node, MappingNode):182 constructor = self.__class__.construct_mapping183 if tag_suffix is None:184 data = constructor(self, node)185 else:186 data = constructor(self, tag_suffix, node)187 if isinstance(data, types.GeneratorType):188 generator = data189 data = next(generator)190 if self.deep_construct:191 for _dummy in generator:192 pass193 else:194 self.state_generators.append(generator)195 return data196 197 def construct_scalar(self, node: Any) -> Any:198 if not isinstance(node, ScalarNode):199 raise ConstructorError(200 None, None, f'expected a scalar node, but found {node.id!s}', node.start_mark,201 )202 return node.value203 204 def construct_sequence(self, node: Any, deep: bool = False) -> Any:205 """deep is True when creating an object/mapping recursively,206 in that case want the underlying elements available during construction207 """208 if not isinstance(node, SequenceNode):209 raise ConstructorError(210 None,211 None,212 f'expected a sequence node, but found {node.id!s}',213 node.start_mark,214 )215 return [self.construct_object(child, deep=deep) for child in node.value]216 217 def construct_mapping(self, node: Any, deep: bool = False) -> Any:218 """deep is True when creating an object/mapping recursively,219 in that case want the underlying elements available during construction220 """221 if not isinstance(node, MappingNode):222 raise ConstructorError(223 None, None, f'expected a mapping node, but found {node.id!s}', node.start_mark,224 )225 total_mapping = self.yaml_base_dict_type()226 if getattr(node, 'merge', None) is not None:227 todo = [(node.merge, False), (node.value, False)]228 else:229 todo = [(node.value, True)]230 for values, check in todo:231 mapping: Dict[Any, Any] = self.yaml_base_dict_type()232 for key_node, value_node in values:233 # keys can be list -> deep234 key = self.construct_object(key_node, deep=True)235 # lists are not hashable, but tuples are236 if not isinstance(key, Hashable):237 if isinstance(key, list):238 key = tuple(key)239 if not isinstance(key, Hashable):240 raise ConstructorError(241 'while constructing a mapping',242 node.start_mark,243 'found unhashable key',244 key_node.start_mark,245 )246 247 value = self.construct_object(value_node, deep=deep)248 if check:249 if self.check_mapping_key(node, key_node, mapping, key, value):250 mapping[key] = value251 else:252 mapping[key] = value253 total_mapping.update(mapping)254 return total_mapping255 256 def check_mapping_key(257 self, node: Any, key_node: Any, mapping: Any, key: Any, value: Any,258 ) -> bool:259 """return True if key is unique"""260 if key in mapping:261 if not self.allow_duplicate_keys:262 mk = mapping.get(key)263 args = [264 'while constructing a mapping',265 node.start_mark,266 f'found duplicate key "{key}" with value "{value}" '267 f'(original value: "{mk}")',268 key_node.start_mark,269 f"""270 To suppress this check see:271 {DUPKEY_URL}272 """,273 """\274 Duplicate keys will become an error in future releases, and are errors275 by default when using the new API.276 """,277 ]278 if self.allow_duplicate_keys is None:279 warnings.warn(DuplicateKeyFutureWarning(*args), stacklevel=1)280 else:281 raise DuplicateKeyError(*args)282 return False283 return True284 285 def check_set_key(self: Any, node: Any, key_node: Any, setting: Any, key: Any) -> None:286 if key in setting:287 if not self.allow_duplicate_keys:288 args = [289 'while constructing a set',290 node.start_mark,291 f'found duplicate key "{key}"',292 key_node.start_mark,293 f"""294 To suppress this check see:295 {DUPKEY_URL}296 """,297 """\298 Duplicate keys will become an error in future releases, and are errors299 by default when using the new API.300 """,301 ]302 if self.allow_duplicate_keys is None:303 warnings.warn(DuplicateKeyFutureWarning(*args), stacklevel=1)304 else:305 raise DuplicateKeyError(*args)306 307 def construct_pairs(self, node: Any, deep: bool = False) -> Any:308 if not isinstance(node, MappingNode):309 raise ConstructorError(310 None, None, f'expected a mapping node, but found {node.id!s}', node.start_mark,311 )312 pairs = []313 for key_node, value_node in node.value:314 key = self.construct_object(key_node, deep=deep)315 value = self.construct_object(value_node, deep=deep)316 pairs.append((key, value))317 return pairs318 319 # ToDo: putting stuff on the class makes it global, consider making this to work on an320 # instance variable once function load is dropped.321 @classmethod322 def add_constructor(cls, tag: Any, constructor: Any) -> Any:323 if isinstance(tag, Tag):324 tag = str(tag)325 if 'yaml_constructors' not in cls.__dict__:326 cls.yaml_constructors = cls.yaml_constructors.copy()327 ret_val = cls.yaml_constructors.get(tag, None)328 cls.yaml_constructors[tag] = constructor329 return ret_val330 331 @classmethod332 def add_multi_constructor(cls, tag_prefix: Any, multi_constructor: Any) -> None:333 if 'yaml_multi_constructors' not in cls.__dict__:334 cls.yaml_multi_constructors = cls.yaml_multi_constructors.copy()335 cls.yaml_multi_constructors[tag_prefix] = multi_constructor336 337 @classmethod338 def add_default_constructor(339 cls, tag: str, method: Any = None, tag_base: str = 'tag:yaml.org,2002:',340 ) -> None:341 if not tag.startswith('tag:'):342 if method is None:343 method = 'construct_yaml_' + tag344 tag = tag_base + tag345 cls.add_constructor(tag, getattr(cls, method))346 347 348class SafeConstructor(BaseConstructor):349 def construct_scalar(self, node: Any) -> Any:350 if isinstance(node, MappingNode):351 for key_node, value_node in node.value:352 if key_node.tag == 'tag:yaml.org,2002:value':353 return self.construct_scalar(value_node)354 return BaseConstructor.construct_scalar(self, node)355 356 def flatten_mapping(self, node: Any) -> Any: # SafeConstructor357 """358 This implements the merge key feature http://yaml.org/type/merge.html359 by inserting keys from the merge dict/list of dicts if not yet360 available in this node361 """362 merge: List[Any] = []363 index = 0364 while index < len(node.value):365 key_node, value_node = node.value[index]366 if key_node.tag == 'tag:yaml.org,2002:merge':367 if merge: # double << key368 if self.allow_duplicate_keys:369 del node.value[index]370 index += 1371 continue372 args = [373 'while constructing a mapping',374 node.start_mark,375 'found duplicate merge key "<<"',376 key_node.start_mark,377 """\378 Duplicate merge keys are never allowed, not even when379 `.allow_duplicate_keys` is set to True380 """,381 ]382 raise DuplicateKeyError(*args)383 del node.value[index]384 if isinstance(value_node, MappingNode):385 self.flatten_mapping(value_node)386 merge.extend(value_node.value)387 elif isinstance(value_node, SequenceNode):388 submerge = []389 for subnode in value_node.value:390 if not isinstance(subnode, MappingNode):391 raise ConstructorError(392 'while constructing a mapping',393 node.start_mark,394 f'expected a mapping for merging, but found {subnode.id!s}',395 subnode.start_mark,396 )397 self.flatten_mapping(subnode)398 submerge.append(subnode.value)399 submerge.reverse()400 for value in submerge:401 merge.extend(value)402 else:403 raise ConstructorError(404 'while constructing a mapping',405 node.start_mark,406 'expected a mapping or list of mappings for merging, '407 f'but found {value_node.id!s}',408 value_node.start_mark,409 )410 elif key_node.tag == 'tag:yaml.org,2002:value':411 key_node.tag = 'tag:yaml.org,2002:str'412 index += 1413 else:414 index += 1415 if bool(merge):416 node.merge = merge # separate merge keys to be able to update without duplicate417 node.value = merge + node.value418 419 def construct_mapping(self, node: Any, deep: bool = False) -> Any:420 """deep is True when creating an object/mapping recursively,421 in that case want the underlying elements available during construction422 """423 if isinstance(node, MappingNode):424 self.flatten_mapping(node)425 return BaseConstructor.construct_mapping(self, node, deep=deep)426 427 def construct_yaml_null(self, node: Any) -> Any:428 self.construct_scalar(node)429 return None430 431 # YAML 1.2 spec doesn't mention yes/no etc any more, 1.1 does432 bool_values = {433 'yes': True,434 'no': False,435 'y': True,436 'n': False,437 'true': True,438 'false': False,439 'on': True,440 'off': False,441 }442 443 def construct_yaml_bool(self, node: Any) -> bool:444 value = self.construct_scalar(node)445 return self.bool_values[value.lower()]446 447 def construct_yaml_int(self, node: Any) -> int:448 value_s = self.construct_scalar(node)449 value_s = value_s.replace('_', "")450 sign = +1451 if value_s[0] == '-':452 sign = -1453 if value_s[0] in '+-':454 value_s = value_s[1:]455 if value_s == '0':456 return 0457 elif value_s.startswith('0b'):458 return sign * int(value_s[2:], 2)459 elif value_s.startswith('0x'):460 return sign * int(value_s[2:], 16)461 elif value_s.startswith('0o'):462 return sign * int(value_s[2:], 8)463 elif self.resolver.processing_version == (1, 1) and value_s[0] == '0':464 return sign * int(value_s, 8)465 elif self.resolver.processing_version == (1, 1) and ':' in value_s:466 digits = [int(part) for part in value_s.split(':')]467 digits.reverse()468 base = 1469 value = 0470 for digit in digits:471 value += digit * base472 base *= 60473 return sign * value474 else:475 return sign * int(value_s)476 477 inf_value = 1e300478 while inf_value != inf_value * inf_value:479 inf_value *= inf_value480 nan_value = -inf_value / inf_value # Trying to make a quiet NaN (like C99).481 482 def construct_yaml_float(self, node: Any) -> float:483 value_so = self.construct_scalar(node)484 value_s = value_so.replace('_', "").lower()485 sign = +1486 if value_s[0] == '-':487 sign = -1488 if value_s[0] in '+-':489 value_s = value_s[1:]490 if value_s == '.inf':491 return sign * self.inf_value492 elif value_s == '.nan':493 return self.nan_value494 elif self.resolver.processing_version != (1, 2) and ':' in value_s:495 digits = [float(part) for part in value_s.split(':')]496 digits.reverse()497 base = 1498 value = 0.0499 for digit in digits:500 value += digit * base501 base *= 60502 return sign * value503 else:504 if self.resolver.processing_version != (1, 2) and 'e' in value_s:505 # value_s is lower case independent of input506 mantissa, exponent = value_s.split('e')507 if '.' not in mantissa:508 warnings.warn(MantissaNoDotYAML1_1Warning(node, value_so), stacklevel=1)509 return sign * float(value_s)510 511 def construct_yaml_binary(self, node: Any) -> Any:512 import base64513 514 try:515 value = self.construct_scalar(node).encode('ascii')516 except UnicodeEncodeError as exc:517 raise ConstructorError(518 None,519 None,520 f'failed to convert base64 data into ascii: {exc!s}',521 node.start_mark,522 )523 try:524 return base64.decodebytes(value)525 except binascii.Error as exc:526 raise ConstructorError(527 None, None, f'failed to decode base64 data: {exc!s}', node.start_mark,528 )529 530 timestamp_regexp = timestamp_regexp # moved to util 0.17.17531 532 def construct_yaml_timestamp(self, node: Any, values: Any = None) -> Any:533 if values is None:534 try:535 match = self.timestamp_regexp.match(node.value)536 except TypeError:537 match = None538 if match is None:539 raise ConstructorError(540 None,541 None,542 f'failed to construct timestamp from "{node.value}"',543 node.start_mark,544 )545 values = match.groupdict()546 return create_timestamp(**values)547 548 def construct_yaml_omap(self, node: Any) -> Any:549 # Note: we do now check for duplicate keys550 omap = ordereddict()551 yield omap552 if not isinstance(node, SequenceNode):553 raise ConstructorError(554 'while constructing an ordered map',555 node.start_mark,556 f'expected a sequence, but found {node.id!s}',557 node.start_mark,558 )559 for subnode in node.value:560 if not isinstance(subnode, MappingNode):561 raise ConstructorError(562 'while constructing an ordered map',563 node.start_mark,564 f'expected a mapping of length 1, but found {subnode.id!s}',565 subnode.start_mark,566 )567 if len(subnode.value) != 1:568 raise ConstructorError(569 'while constructing an ordered map',570 node.start_mark,571 f'expected a single mapping item, but found {len(subnode.value):d} items',572 subnode.start_mark,573 )574 key_node, value_node = subnode.value[0]575 key = self.construct_object(key_node)576 assert key not in omap577 value = self.construct_object(value_node)578 omap[key] = value579 580 def construct_yaml_pairs(self, node: Any) -> Any:581 # Note: the same code as `construct_yaml_omap`.582 pairs: List[Any] = []583 yield pairs584 if not isinstance(node, SequenceNode):585 raise ConstructorError(586 'while constructing pairs',587 node.start_mark,588 f'expected a sequence, but found {node.id!s}',589 node.start_mark,590 )591 for subnode in node.value:592 if not isinstance(subnode, MappingNode):593 raise ConstructorError(594 'while constructing pairs',595 node.start_mark,596 f'expected a mapping of length 1, but found {subnode.id!s}',597 subnode.start_mark,598 )599 if len(subnode.value) != 1:600 raise ConstructorError(601 'while constructing pairs',602 node.start_mark,603 f'expected a single mapping item, but found {len(subnode.value):d} items',604 subnode.start_mark,605 )606 key_node, value_node = subnode.value[0]607 key = self.construct_object(key_node)608 value = self.construct_object(value_node)609 pairs.append((key, value))610 611 def construct_yaml_set(self, node: Any) -> Any:612 data: Set[Any] = set()613 yield data614 value = self.construct_mapping(node)615 data.update(value)616 617 def construct_yaml_str(self, node: Any) -> Any:618 value = self.construct_scalar(node)619 return value620 621 def construct_yaml_seq(self, node: Any) -> Any:622 data: List[Any] = self.yaml_base_list_type()623 yield data624 data.extend(self.construct_sequence(node))625 626 def construct_yaml_map(self, node: Any) -> Any:627 data: Dict[Any, Any] = self.yaml_base_dict_type()628 yield data629 value = self.construct_mapping(node)630 data.update(value)631 632 def construct_yaml_object(self, node: Any, cls: Any) -> Any:633 data = cls.__new__(cls)634 yield data635 if hasattr(data, '__setstate__'):636 state = self.construct_mapping(node, deep=True)637 data.__setstate__(state)638 else:639 state = self.construct_mapping(node)640 data.__dict__.update(state)641 642 def construct_undefined(self, node: Any) -> None:643 raise ConstructorError(644 None,645 None,646 f'could not determine a constructor for the tag {node.tag!r}',647 node.start_mark,648 )649 650 651for tag in 'null bool int float binary timestamp omap pairs set str seq map'.split():652 SafeConstructor.add_default_constructor(tag)653 654SafeConstructor.add_constructor(None, SafeConstructor.construct_undefined)655 656 657class Constructor(SafeConstructor):658 def construct_python_str(self, node: Any) -> Any:659 return self.construct_scalar(node)660 661 def construct_python_unicode(self, node: Any) -> Any:662 return self.construct_scalar(node)663 664 def construct_python_bytes(self, node: Any) -> Any:665 import base64666 667 try:668 value = self.construct_scalar(node).encode('ascii')669 except UnicodeEncodeError as exc:670 raise ConstructorError(671 None,672 None,673 f'failed to convert base64 data into ascii: {exc!s}',674 node.start_mark,675 )676 try:677 return base64.decodebytes(value)678 except binascii.Error as exc:679 raise ConstructorError(680 None, None, f'failed to decode base64 data: {exc!s}', node.start_mark,681 )682 683 def construct_python_long(self, node: Any) -> int:684 val = self.construct_yaml_int(node)685 return val686 687 def construct_python_complex(self, node: Any) -> Any:688 return complex(self.construct_scalar(node))689 690 def construct_python_tuple(self, node: Any) -> Any:691 return tuple(self.construct_sequence(node))692 693 def find_python_module(self, name: Any, mark: Any) -> Any:694 if not name:695 raise ConstructorError(696 'while constructing a Python module',697 mark,698 'expected non-empty name appended to the tag',699 mark,700 )701 try:702 __import__(name)703 except ImportError as exc:704 raise ConstructorError(705 'while constructing a Python module',706 mark,707 f'cannot find module {name!r} ({exc!s})',708 mark,709 )710 return sys.modules[name]711 712 def find_python_name(self, name: Any, mark: Any) -> Any:713 if not name:714 raise ConstructorError(715 'while constructing a Python object',716 mark,717 'expected non-empty name appended to the tag',718 mark,719 )720 if '.' in name:721 lname = name.split('.')722 lmodule_name = lname723 lobject_name: List[Any] = []724 while len(lmodule_name) > 1:725 lobject_name.insert(0, lmodule_name.pop())726 module_name = '.'.join(lmodule_name)727 try:728 __import__(module_name)729 # object_name = '.'.join(object_name)730 break731 except ImportError:732 continue733 else:734 module_name = builtins_module735 lobject_name = [name]736 try:737 __import__(module_name)738 except ImportError as exc:739 raise ConstructorError(740 'while constructing a Python object',741 mark,742 f'cannot find module {module_name!r} ({exc!s})',743 mark,744 )745 module = sys.modules[module_name]746 object_name = '.'.join(lobject_name)747 obj = module748 while lobject_name:749 if not hasattr(obj, lobject_name[0]):750 751 raise ConstructorError(752 'while constructing a Python object',753 mark,754 f'cannot find {object_name!r} in the module {module.__name__!r}',755 mark,756 )757 obj = getattr(obj, lobject_name.pop(0))758 return obj759 760 def construct_python_name(self, suffix: Any, node: Any) -> Any:761 value = self.construct_scalar(node)762 if value:763 raise ConstructorError(764 'while constructing a Python name',765 node.start_mark,766 f'expected the empty value, but found {value!r}',767 node.start_mark,768 )769 return self.find_python_name(suffix, node.start_mark)770 771 def construct_python_module(self, suffix: Any, node: Any) -> Any:772 value = self.construct_scalar(node)773 if value:774 raise ConstructorError(775 'while constructing a Python module',776 node.start_mark,777 f'expected the empty value, but found {value!r}',778 node.start_mark,779 )780 return self.find_python_module(suffix, node.start_mark)781 782 def make_python_instance(783 self, suffix: Any, node: Any, args: Any = None, kwds: Any = None, newobj: bool = False,784 ) -> Any:785 if not args:786 args = []787 if not kwds:788 kwds = {}789 cls = self.find_python_name(suffix, node.start_mark)790 if newobj and isinstance(cls, type):791 return cls.__new__(cls, *args, **kwds)792 else:793 return cls(*args, **kwds)794 795 def set_python_instance_state(self, instance: Any, state: Any) -> None:796 if hasattr(instance, '__setstate__'):797 instance.__setstate__(state)798 else:799 slotstate: Dict[Any, Any] = {}800 if isinstance(state, tuple) and len(state) == 2:801 state, slotstate = state802 if hasattr(instance, '__dict__'):803 instance.__dict__.update(state)804 elif state:805 slotstate.update(state)806 for key, value in slotstate.items():807 setattr(instance, key, value)808 809 def construct_python_object(self, suffix: Any, node: Any) -> Any:810 # Format:811 # !!python/object:module.name { ... state ... }812 instance = self.make_python_instance(suffix, node, newobj=True)813 self.recursive_objects[node] = instance814 yield instance815 deep = hasattr(instance, '__setstate__')816 state = self.construct_mapping(node, deep=deep)817 self.set_python_instance_state(instance, state)818 819 def construct_python_object_apply(820 self, suffix: Any, node: Any, newobj: bool = False,821 ) -> Any:822 # Format:823 # !!python/object/apply # (or !!python/object/new)824 # args: [ ... arguments ... ]825 # kwds: { ... keywords ... }826 # state: ... state ...827 # listitems: [ ... listitems ... ]828 # dictitems: { ... dictitems ... }829 # or short format:830 # !!python/object/apply [ ... arguments ... ]831 # The difference between !!python/object/apply and !!python/object/new832 # is how an object is created, check make_python_instance for details.833 if isinstance(node, SequenceNode):834 args = self.construct_sequence(node, deep=True)835 kwds: Dict[Any, Any] = {}836 state: Dict[Any, Any] = {}837 listitems: List[Any] = []838 dictitems: Dict[Any, Any] = {}839 else:840 value = self.construct_mapping(node, deep=True)841 args = value.get('args', [])842 kwds = value.get('kwds', {})843 state = value.get('state', {})844 listitems = value.get('listitems', [])845 dictitems = value.get('dictitems', {})846 instance = self.make_python_instance(suffix, node, args, kwds, newobj)847 if bool(state):848 self.set_python_instance_state(instance, state)849 if bool(listitems):850 instance.extend(listitems)851 if bool(dictitems):852 for key in dictitems:853 instance[key] = dictitems[key]854 return instance855 856 def construct_python_object_new(self, suffix: Any, node: Any) -> Any:857 return self.construct_python_object_apply(suffix, node, newobj=True)858 859 @classmethod860 def add_default_constructor(861 cls, tag: str, method: Any = None, tag_base: str = 'tag:yaml.org,2002:python/',862 ) -> None:863 if not tag.startswith('tag:'):864 if method is None:865 method = 'construct_yaml_' + tag866 tag = tag_base + tag867 cls.add_constructor(tag, getattr(cls, method))868 869 870Constructor.add_constructor('tag:yaml.org,2002:python/none', Constructor.construct_yaml_null)871 872Constructor.add_constructor('tag:yaml.org,2002:python/bool', Constructor.construct_yaml_bool)873 874Constructor.add_constructor('tag:yaml.org,2002:python/str', Constructor.construct_python_str)875 876Constructor.add_constructor(877 'tag:yaml.org,2002:python/unicode', Constructor.construct_python_unicode,878)879 880Constructor.add_constructor(881 'tag:yaml.org,2002:python/bytes', Constructor.construct_python_bytes,882)883 884Constructor.add_constructor('tag:yaml.org,2002:python/int', Constructor.construct_yaml_int)885 886Constructor.add_constructor('tag:yaml.org,2002:python/long', Constructor.construct_python_long)887 888Constructor.add_constructor('tag:yaml.org,2002:python/float', Constructor.construct_yaml_float)889 890Constructor.add_constructor(891 'tag:yaml.org,2002:python/complex', Constructor.construct_python_complex,892)893 894Constructor.add_constructor('tag:yaml.org,2002:python/list', Constructor.construct_yaml_seq)895 896Constructor.add_constructor(897 'tag:yaml.org,2002:python/tuple', Constructor.construct_python_tuple,898)899# for tag in 'bool str unicode bytes int long float complex tuple'.split():900# Constructor.add_default_constructor(tag)901 902Constructor.add_constructor('tag:yaml.org,2002:python/dict', Constructor.construct_yaml_map)903 904Constructor.add_multi_constructor(905 'tag:yaml.org,2002:python/name:', Constructor.construct_python_name,906)907 908Constructor.add_multi_constructor(909 'tag:yaml.org,2002:python/module:', Constructor.construct_python_module,910)911 912Constructor.add_multi_constructor(913 'tag:yaml.org,2002:python/object:', Constructor.construct_python_object,914)915 916Constructor.add_multi_constructor(917 'tag:yaml.org,2002:python/object/apply:', Constructor.construct_python_object_apply,918)919 920Constructor.add_multi_constructor(921 'tag:yaml.org,2002:python/object/new:', Constructor.construct_python_object_new,922)923 924 925class RoundTripConstructor(SafeConstructor):926 """need to store the comments on the node itself,927 as well as on the items928 """929 930 def comment(self, idx: Any) -> Any:931 assert self.loader.comment_handling is not None932 x = self.scanner.comments[idx]933 x.set_assigned()934 return x935 936 def comments(self, list_of_comments: Any, idx: Optional[Any] = None) -> Any:937 # hand in the comment and optional pre, eol, post segment938 if list_of_comments is None:939 return []940 if idx is not None:941 if list_of_comments[idx] is None:942 return []943 list_of_comments = list_of_comments[idx]944 for x in list_of_comments:945 yield self.comment(x)946 947 def construct_scalar(self, node: Any) -> Any:948 if not isinstance(node, ScalarNode):949 raise ConstructorError(950 None, None, f'expected a scalar node, but found {node.id!s}', node.start_mark,951 )952 953 if node.style == '|' and isinstance(node.value, str):954 lss = LiteralScalarString(node.value, anchor=node.anchor)955 if self.loader and self.loader.comment_handling is None:956 if node.comment and node.comment[1]:957 lss.comment = node.comment[1][0] # type: ignore958 else:959 # NEWCMNT960 if node.comment is not None and node.comment[1]:961 # nprintf('>>>>nc1', node.comment)962 # EOL comment after |963 lss.comment = self.comment(node.comment[1][0]) # type: ignore964 return lss965 if node.style == '>' and isinstance(node.value, str):966 fold_positions: List[int] = []967 idx = -1968 while True:969 idx = node.value.find('\a', idx + 1)970 if idx < 0:971 break972 fold_positions.append(idx - len(fold_positions))973 fss = FoldedScalarString(node.value.replace('\a', ''), anchor=node.anchor)974 if self.loader and self.loader.comment_handling is None:975 if node.comment and node.comment[1]:976 fss.comment = node.comment[1][0] # type: ignore977 else:978 # NEWCMNT979 if node.comment is not None and node.comment[1]:980 # nprintf('>>>>nc2', node.comment)981 # EOL comment after >982 fss.comment = self.comment(node.comment[1][0]) # type: ignore983 if fold_positions:984 fss.fold_pos = fold_positions # type: ignore985 return fss986 elif bool(self._preserve_quotes) and isinstance(node.value, str):987 if node.style == "'":988 return SingleQuotedScalarString(node.value, anchor=node.anchor)989 if node.style == '"':990 return DoubleQuotedScalarString(node.value, anchor=node.anchor)991 # if node.ctag:992 # data2 = TaggedScalar()993 # data2.value = node.value994 # data2.style = node.style995 # data2.yaml_set_ctag(node.ctag)996 # if node.anchor:997 # from ruamel.yaml.serializer import templated_id998 999 # if not templated_id(node.anchor):1000 # data2.yaml_set_anchor(node.anchor, always_dump=True)1001 # return data21002 if node.anchor:1003 return PlainScalarString(node.value, anchor=node.anchor)1004 return node.value1005 1006 def construct_yaml_int(self, node: Any) -> Any:1007 width: Any = None1008 value_su = self.construct_scalar(node)1009 try:1010 sx = value_su.rstrip('_')1011 underscore: Any = [len(sx) - sx.rindex('_') - 1, False, False]1012 except ValueError:1013 underscore = None1014 except IndexError:1015 underscore = None1016 value_s = value_su.replace('_', "")1017 sign = +11018 if value_s[0] in '+-':1019 if value_s[0] == '-':1020 sign = -11021 value_s = value_s[1:]1022 if value_s.startswith('0b'):1023 if self.resolver.processing_version > (1, 1) and value_s[2] == '0':1024 width = len(value_s[2:])1025 if underscore is not None:1026 underscore[1] = value_su[2] == '_'1027 underscore[2] = len(value_su[2:]) > 1 and value_su[-1] == '_'1028 return BinaryInt(1029 sign * int(value_s[2:], 2),1030 width=width,1031 underscore=underscore,1032 anchor=node.anchor,1033 )1034 elif value_s.startswith('0x'):1035 # default to lower-case if no a-fA-F in string1036 if self.resolver.processing_version > (1, 1) and value_s[2] == '0':1037 width = len(value_s[2:])1038 hex_fun: Any = HexInt1039 for ch in value_s[2:]:1040 if ch in 'ABCDEF': # first non-digit is capital1041 hex_fun = HexCapsInt1042 break1043 if ch in 'abcdef':1044 break1045 if underscore is not None:1046 underscore[1] = value_su[2] == '_'1047 underscore[2] = len(value_su[2:]) > 1 and value_su[-1] == '_'1048 return hex_fun(1049 sign * int(value_s[2:], 16),1050 width=width,1051 underscore=underscore,1052 anchor=node.anchor,1053 )1054 elif value_s.startswith('0o'):1055 if self.resolver.processing_version > (1, 1) and value_s[2] == '0':1056 width = len(value_s[2:])1057 if underscore is not None:1058 underscore[1] = value_su[2] == '_'1059 underscore[2] = len(value_su[2:]) > 1 and value_su[-1] == '_'1060 return OctalInt(1061 sign * int(value_s[2:], 8),1062 width=width,1063 underscore=underscore,1064 anchor=node.anchor,1065 )1066 elif (1067 self.resolver.processing_version != (1, 2)1068 and len(value_s) > 11069 and value_s[0] == '0'1070 ):1071 return OctalInt(1072 sign * int(value_s, 8), width=width, underscore=underscore, anchor=node.anchor,1073 )1074 elif self.resolver.processing_version != (1, 2) and ':' in value_s:1075 digits = [int(part) for part in value_s.split(':')]1076 digits.reverse()1077 base = 11078 value = 01079 for digit in digits:1080 value += digit * base1081 base *= 601082 return sign * value1083 elif self.resolver.processing_version > (1, 1) and value_s[0] == '0':1084 # not an octal, an integer with leading zero(s)1085 if underscore is not None:1086 # cannot have a leading underscore1087 underscore[2] = len(value_su) > 1 and value_su[-1] == '_'1088 return ScalarInt(1089 sign * int(value_s),1090 width=len(value_s),1091 underscore=underscore,1092 anchor=node.anchor,1093 )1094 elif underscore:1095 # cannot have a leading underscore1096 underscore[2] = len(value_su) > 1 and value_su[-1] == '_'1097 return ScalarInt(1098 sign * int(value_s), width=None, underscore=underscore, anchor=node.anchor,1099 )1100 elif node.anchor:1101 return ScalarInt(sign * int(value_s), width=None, anchor=node.anchor)1102 else:1103 return sign * int(value_s)1104 1105 def construct_yaml_float(self, node: Any) -> Any:1106 def leading_zeros(v: Any) -> int:1107 lead0 = 01108 idx = 01109 while idx < len(v) and v[idx] in '0.':1110 if v[idx] == '0':1111 lead0 += 11112 idx += 11113 return lead01114 1115 # underscore = None1116 m_sign: Any = False1117 value_so = self.construct_scalar(node)1118 value_s = value_so.replace('_', "").lower()1119 sign = +11120 if value_s[0] == '-':1121 sign = -11122 if value_s[0] in '+-':1123 m_sign = value_s[0]1124 value_s = value_s[1:]1125 if value_s == '.inf':1126 return sign * self.inf_value1127 if value_s == '.nan':1128 return self.nan_value1129 if self.resolver.processing_version != (1, 2) and ':' in value_s:1130 digits = [float(part) for part in value_s.split(':')]1131 digits.reverse()1132 base = 11133 value = 0.01134 for digit in digits:1135 value += digit * base1136 base *= 601137 return sign * value1138 if 'e' in value_s:1139 try:1140 mantissa, exponent = value_so.split('e')1141 exp = 'e'1142 except ValueError:1143 mantissa, exponent = value_so.split('E')1144 exp = 'E'1145 if self.resolver.processing_version != (1, 2):1146 # value_s is lower case independent of input1147 if '.' not in mantissa:1148 warnings.warn(MantissaNoDotYAML1_1Warning(node, value_so), stacklevel=1)1149 lead0 = leading_zeros(mantissa)1150 width = len(mantissa)1151 prec = mantissa.find('.')1152 if m_sign:1153 width -= 11154 e_width = len(exponent)1155 e_sign = exponent[0] in '+-'1156 # nprint('sf', width, prec, m_sign, exp, e_width, e_sign)1157 return ScalarFloat(1158 sign * float(value_s),1159 width=width,1160 prec=prec,1161 m_sign=m_sign,1162 m_lead0=lead0,1163 exp=exp,1164 e_width=e_width,1165 e_sign=e_sign,1166 anchor=node.anchor,1167 )1168 width = len(value_so)1169 # you can't use index, !!float 42 would be a float without a dot1170 prec = value_so.find('.')1171 lead0 = leading_zeros(value_so)1172 return ScalarFloat(1173 sign * float(value_s),1174 width=width,1175 prec=prec,1176 m_sign=m_sign,1177 m_lead0=lead0,1178 anchor=node.anchor,1179 )1180 1181 def construct_yaml_str(self, node: Any) -> Any:1182 if node.ctag.handle:1183 value = self.construct_unknown(node)1184 else:1185 value = self.construct_scalar(node)1186 if isinstance(value, ScalarString):1187 return value1188 return value1189 1190 def construct_rt_sequence(self, node: Any, seqtyp: Any, deep: bool = False) -> Any:1191 if not isinstance(node, SequenceNode):1192 raise ConstructorError(1193 None,1194 None,1195 f'expected a sequence node, but found {node.id!s}',1196 node.start_mark,1197 )1198 ret_val = []1199 if self.loader and self.loader.comment_handling is None:1200 if node.comment: