codekingpro/portable-devtools
114k
1 2from __future__ import annotations3 4import sys5import os6import warnings7import glob8from importlib import import_module9 10 11import ruamel.yaml12from ruamel.yaml.error import UnsafeLoaderWarning, YAMLError # NOQA13 14from ruamel.yaml.tokens import * # NOQA15from ruamel.yaml.events import * # NOQA16from ruamel.yaml.nodes import * # NOQA17 18from ruamel.yaml.loader import BaseLoader, SafeLoader, Loader, RoundTripLoader # NOQA19from ruamel.yaml.dumper import BaseDumper, SafeDumper, Dumper, RoundTripDumper # NOQA20from ruamel.yaml.compat import StringIO, BytesIO, with_metaclass, nprint, nprintf # NOQA21from ruamel.yaml.resolver import VersionedResolver, Resolver # NOQA22from ruamel.yaml.representer import (23 BaseRepresenter,24 SafeRepresenter,25 Representer,26 RoundTripRepresenter,27)28from ruamel.yaml.constructor import (29 BaseConstructor,30 SafeConstructor,31 Constructor,32 RoundTripConstructor,33)34from ruamel.yaml.loader import Loader as UnsafeLoader # NOQA35from ruamel.yaml.comments import CommentedMap, CommentedSeq, C_PRE36from ruamel.yaml.docinfo import DocInfo, version, Version37 38from typing import List, Set, Dict, Tuple, Union, Any, Callable, Optional, Text, Type # NOQA39if False: # MYPY40 from ruamel.yaml.compat import StreamType, StreamTextType, VersionType # NOQA41 from types import TracebackType42 from pathlib import Path43 44try:45 from _ruamel_yaml import CParser, CEmitter # type: ignore46except: # NOQA47 CParser = CEmitter = None48 49# import io50 51 52# YAML is an acronym, i.e. spoken: rhymes with "camel". And thus a53# subset of abbreviations, which should be all caps according to PEP854 55 56class YAML:57 def __init__(58 self: Any,59 *,60 typ: Optional[Union[List[Text], Text]] = None,61 pure: Any = False,62 output: Any = None,63 plug_ins: Any = None,64 ) -> None: # input=None,65 """66 typ: 'rt'/None -> RoundTripLoader/RoundTripDumper, (default)67 'safe' -> SafeLoader/SafeDumper,68 'unsafe' -> normal/unsafe Loader/Dumper (pending deprecation)69 'full' -> full Dumper only, including python built-ins that are70 potentially unsafe to load71 'base' -> baseloader72 pure: if True only use Python modules73 input/output: needed to work as context manager74 plug_ins: a list of plug-in files75 """76 77 self.typ = ['rt'] if typ is None else (typ if isinstance(typ, list) else [typ])78 self.pure = pure79 80 # self._input = input81 self._output = output82 self._context_manager: Any = None83 84 self.plug_ins: List[Any] = []85 for pu in ([] if plug_ins is None else plug_ins) + self.official_plug_ins():86 file_name = pu.replace(os.sep, '.')87 self.plug_ins.append(import_module(file_name))88 self.Resolver: Any = ruamel.yaml.resolver.VersionedResolver89 self.allow_unicode = True90 self.Reader: Any = None91 self.Representer: Any = None92 self.Constructor: Any = None93 self.Scanner: Any = None94 self.Serializer: Any = None95 self.default_flow_style: Any = None96 self.comment_handling = None97 self.max_depth = 098 typ_found = 199 setup_rt = False100 if 'rt' in self.typ:101 setup_rt = True102 elif 'safe' in self.typ:103 self.Emitter = (104 ruamel.yaml.emitter.Emitter if pure or CEmitter is None else CEmitter105 )106 self.Representer = ruamel.yaml.representer.SafeRepresenter107 self.Parser = ruamel.yaml.parser.Parser if pure or CParser is None else CParser108 self.Composer = ruamel.yaml.composer.Composer109 self.Constructor = ruamel.yaml.constructor.SafeConstructor110 elif 'base' in self.typ:111 self.Emitter = ruamel.yaml.emitter.Emitter112 self.Representer = ruamel.yaml.representer.BaseRepresenter113 self.Parser = ruamel.yaml.parser.Parser if pure or CParser is None else CParser114 self.Composer = ruamel.yaml.composer.Composer115 self.Constructor = ruamel.yaml.constructor.BaseConstructor116 elif 'unsafe' in self.typ:117 warnings.warn(118 "\nyou should no longer specify 'unsafe'.\nFor **dumping only** use yaml=YAML(typ='full')\n", # NOQA119 PendingDeprecationWarning,120 stacklevel=2,121 )122 self.Emitter = (123 ruamel.yaml.emitter.Emitter if pure or CEmitter is None else CEmitter124 )125 self.Representer = ruamel.yaml.representer.Representer126 self.Parser = ruamel.yaml.parser.Parser if pure or CParser is None else CParser127 self.Composer = ruamel.yaml.composer.Composer128 self.Constructor = ruamel.yaml.constructor.Constructor129 elif 'full' in self.typ:130 self.Emitter = (131 ruamel.yaml.emitter.Emitter if pure or CEmitter is None else CEmitter132 )133 self.Representer = ruamel.yaml.representer.Representer134 self.Parser = ruamel.yaml.parser.Parser if pure or CParser is None else CParser135 # self.Composer = ruamel.yaml.composer.Composer136 # self.Constructor = ruamel.yaml.constructor.Constructor137 elif 'rtsc' in self.typ:138 self.default_flow_style = False139 # no optimized rt-dumper yet140 self.Emitter = ruamel.yaml.emitter.RoundTripEmitter141 self.Serializer = ruamel.yaml.serializer.Serializer142 self.Representer = ruamel.yaml.representer.RoundTripRepresenter143 self.Scanner = ruamel.yaml.scanner.RoundTripScannerSC144 # no optimized rt-parser yet145 self.Parser = ruamel.yaml.parser.RoundTripParserSC146 self.Composer = ruamel.yaml.composer.Composer147 self.Constructor = ruamel.yaml.constructor.RoundTripConstructor148 self.comment_handling = C_PRE149 else:150 setup_rt = True151 typ_found = 0152 if setup_rt:153 self.default_flow_style = False154 # no optimized rt-dumper yet155 self.Emitter = ruamel.yaml.emitter.RoundTripEmitter156 self.Serializer = ruamel.yaml.serializer.Serializer157 self.Representer = ruamel.yaml.representer.RoundTripRepresenter158 self.Scanner = ruamel.yaml.scanner.RoundTripScanner159 # no optimized rt-parser yet160 self.Parser = ruamel.yaml.parser.RoundTripParser161 self.Composer = ruamel.yaml.composer.Composer162 self.Constructor = ruamel.yaml.constructor.RoundTripConstructor163 del setup_rt164 self.stream = None165 self.canonical = None166 self.old_indent = None167 self.width: Union[int, None] = None168 self.line_break = None169 170 self.map_indent: Union[int, None] = None171 self.sequence_indent: Union[int, None] = None172 self.sequence_dash_offset: int = 0173 self.compact_seq_seq = None174 self.compact_seq_map = None175 self.sort_base_mapping_type_on_output = None # default: sort176 177 self.top_level_colon_align = None178 self.prefix_colon = None179 self._version: Optional[Any] = None180 self.preserve_quotes: Optional[bool] = None181 self.allow_duplicate_keys = False # duplicate keys in map, set182 self.encoding = 'utf-8'183 self.explicit_start: Union[bool, None] = None184 self.explicit_end: Union[bool, None] = None185 self._tags = None186 self.doc_infos: List[DocInfo] = []187 self.default_style = None188 self.top_level_block_style_scalar_no_indent_error_1_1 = False189 # directives end indicator with single scalar document190 self.scalar_after_indicator: Optional[bool] = None191 # [a, b: 1, c: {d: 2}] vs. [a, {b: 1}, {c: {d: 2}}]192 self.brace_single_entry_mapping_in_flow_sequence = False193 for module in self.plug_ins:194 if getattr(module, 'typ', None) in self.typ:195 typ_found += 1196 module.init_typ(self)197 break198 if typ_found == 0:199 raise NotImplementedError(200 f'typ "{self.typ}" not recognised (need to install plug-in?)',201 )202 203 @property204 def reader(self) -> Any:205 try:206 return self._reader # type: ignore207 except AttributeError:208 self._reader = self.Reader(None, loader=self)209 return self._reader210 211 @property212 def scanner(self) -> Any:213 try:214 return self._scanner # type: ignore215 except AttributeError:216 if self.Scanner is None:217 raise218 self._scanner = self.Scanner(loader=self)219 return self._scanner220 221 @property222 def parser(self) -> Any:223 attr = '_' + sys._getframe().f_code.co_name224 if not hasattr(self, attr):225 if self.Parser is not CParser:226 setattr(self, attr, self.Parser(loader=self))227 else:228 if getattr(self, '_stream', None) is None:229 # wait for the stream230 return None231 else:232 # if not hasattr(self._stream, 'read') and hasattr(self._stream, 'open'):233 # # pathlib.Path() instance234 # setattr(self, attr, CParser(self._stream))235 # else:236 setattr(self, attr, CParser(self._stream))237 # self._parser = self._composer = self238 # nprint('scanner', self.loader.scanner)239 240 return getattr(self, attr)241 242 @property243 def composer(self) -> Any:244 attr = '_' + sys._getframe().f_code.co_name245 if not hasattr(self, attr):246 setattr(self, attr, self.Composer(loader=self))247 return getattr(self, attr)248 249 @property250 def constructor(self) -> Any:251 attr = '_' + sys._getframe().f_code.co_name252 if not hasattr(self, attr):253 if self.Constructor is None:254 if 'full' in self.typ:255 raise YAMLError(256 "\nyou can only use yaml=YAML(typ='full') for dumping\n", # NOQA257 )258 cnst = self.Constructor(preserve_quotes=self.preserve_quotes, loader=self) # type: ignore # NOQA259 cnst.allow_duplicate_keys = self.allow_duplicate_keys260 setattr(self, attr, cnst)261 return getattr(self, attr)262 263 @property264 def resolver(self) -> Any:265 try:266 rslvr = self._resolver # type: ignore267 except AttributeError:268 rslvr = None269 if rslvr is None or rslvr._loader_version != self.version:270 rslvr = self._resolver = self.Resolver(version=self.version, loader=self)271 return rslvr272 273 @property274 def emitter(self) -> Any:275 attr = '_' + sys._getframe().f_code.co_name276 if not hasattr(self, attr):277 if self.Emitter is not CEmitter:278 _emitter = self.Emitter(279 None,280 canonical=self.canonical,281 indent=self.old_indent,282 width=self.width,283 allow_unicode=self.allow_unicode,284 line_break=self.line_break,285 prefix_colon=self.prefix_colon,286 brace_single_entry_mapping_in_flow_sequence=self.brace_single_entry_mapping_in_flow_sequence, # NOQA287 dumper=self,288 )289 setattr(self, attr, _emitter)290 if self.map_indent is not None:291 _emitter.best_map_indent = self.map_indent292 if self.sequence_indent is not None:293 _emitter.best_sequence_indent = self.sequence_indent294 if self.sequence_dash_offset is not None:295 _emitter.sequence_dash_offset = self.sequence_dash_offset296 # _emitter.block_seq_indent = self.sequence_dash_offset297 if self.compact_seq_seq is not None:298 _emitter.compact_seq_seq = self.compact_seq_seq299 if self.compact_seq_map is not None:300 _emitter.compact_seq_map = self.compact_seq_map301 else:302 if getattr(self, '_stream', None) is None:303 # wait for the stream304 return None305 return None306 return getattr(self, attr)307 308 @property309 def serializer(self) -> Any:310 attr = '_' + sys._getframe().f_code.co_name311 if not hasattr(self, attr):312 setattr(313 self,314 attr,315 self.Serializer(316 encoding=self.encoding,317 explicit_start=self.explicit_start,318 explicit_end=self.explicit_end,319 version=self.version,320 tags=self.tags,321 dumper=self,322 ),323 )324 return getattr(self, attr)325 326 @property327 def representer(self) -> Any:328 attr = '_' + sys._getframe().f_code.co_name329 if not hasattr(self, attr):330 repres = self.Representer(331 default_style=self.default_style,332 default_flow_style=self.default_flow_style,333 dumper=self,334 )335 if self.sort_base_mapping_type_on_output is not None:336 repres.sort_base_mapping_type_on_output = self.sort_base_mapping_type_on_output337 setattr(self, attr, repres)338 return getattr(self, attr)339 340 def scan(self, stream: StreamTextType) -> Any:341 """342 Scan a YAML stream and produce scanning tokens.343 """344 if not hasattr(stream, 'read') and hasattr(stream, 'open'):345 # pathlib.Path() instance346 with stream.open('rb') as fp:347 return self.scan(fp)348 self.doc_infos.append(DocInfo(requested_version=version(self.version)))349 self.tags = {}350 _, parser = self.get_constructor_parser(stream)351 try:352 while self.scanner.check_token():353 yield self.scanner.get_token()354 finally:355 parser.dispose()356 for comp in ('reader', 'scanner'):357 try:358 getattr(getattr(self, '_' + comp), f'reset_{comp}')()359 except AttributeError:360 pass361 362 def parse(self, stream: StreamTextType) -> Any:363 """364 Parse a YAML stream and produce parsing events.365 """366 if not hasattr(stream, 'read') and hasattr(stream, 'open'):367 # pathlib.Path() instance368 with stream.open('rb') as fp:369 return self.parse(fp)370 self.doc_infos.append(DocInfo(requested_version=version(self.version)))371 self.tags = {}372 _, parser = self.get_constructor_parser(stream)373 try:374 while parser.check_event():375 yield parser.get_event()376 finally:377 parser.dispose()378 for comp in ('reader', 'scanner'):379 try:380 getattr(getattr(self, '_' + comp), f'reset_{comp}')()381 except AttributeError:382 pass383 384 def compose(self, stream: Union[Path, StreamTextType]) -> Any:385 """386 Parse the first YAML document in a stream387 and produce the corresponding representation tree.388 """389 if not hasattr(stream, 'read') and hasattr(stream, 'open'):390 # pathlib.Path() instance391 with stream.open('rb') as fp:392 return self.compose(fp)393 self.doc_infos.append(DocInfo(requested_version=version(self.version)))394 self.tags = {}395 constructor, parser = self.get_constructor_parser(stream)396 try:397 return constructor.composer.get_single_node()398 finally:399 parser.dispose()400 for comp in ('reader', 'scanner'):401 try:402 getattr(getattr(self, '_' + comp), f'reset_{comp}')()403 except AttributeError:404 pass405 406 def compose_all(self, stream: Union[Path, StreamTextType]) -> Any:407 """408 Parse all YAML documents in a stream409 and produce corresponding representation trees.410 """411 self.doc_infos.append(DocInfo(requested_version=version(self.version)))412 self.tags = {}413 constructor, parser = self.get_constructor_parser(stream)414 try:415 while constructor.composer.check_node():416 yield constructor.composer.get_node()417 finally:418 parser.dispose()419 for comp in ('reader', 'scanner'):420 try:421 getattr(getattr(self, '_' + comp), f'reset_{comp}')()422 except AttributeError:423 pass424 425 # separate output resolver?426 427 # def load(self, stream=None):428 # if self._context_manager:429 # if not self._input:430 # raise TypeError("Missing input stream while dumping from context manager")431 # for data in self._context_manager.load():432 # yield data433 # return434 # if stream is None:435 # raise TypeError("Need a stream argument when not loading from context manager")436 # return self.load_one(stream)437 438 def load(self, stream: Union[Path, StreamTextType]) -> Any:439 """440 at this point you either have the non-pure Parser (which has its own reader and441 scanner) or you have the pure Parser.442 If the pure Parser is set, then set the Reader and Scanner, if not already set.443 If either the Scanner or Reader are set, you cannot use the non-pure Parser,444 so reset it to the pure parser and set the Reader resp. Scanner if necessary445 """446 if not hasattr(stream, 'read') and hasattr(stream, 'open'):447 # pathlib.Path() instance448 with stream.open('rb') as fp:449 return self.load(fp)450 self.doc_infos.append(DocInfo(requested_version=version(self.version)))451 self.tags = {}452 constructor, parser = self.get_constructor_parser(stream)453 try:454 return constructor.get_single_data()455 finally:456 parser.dispose()457 for comp in ('reader', 'scanner'):458 try:459 getattr(getattr(self, '_' + comp), f'reset_{comp}')()460 except AttributeError:461 pass462 463 def load_all(self, stream: Union[Path, StreamTextType]) -> Any: # *, skip=None):464 if not hasattr(stream, 'read') and hasattr(stream, 'open'):465 # pathlib.Path() instance466 with stream.open('r') as fp:467 for d in self.load_all(fp):468 yield d469 return470 # if skip is None:471 # skip = []472 # elif isinstance(skip, int):473 # skip = [skip]474 self.doc_infos.append(DocInfo(requested_version=version(self.version)))475 self.tags = {}476 constructor, parser = self.get_constructor_parser(stream)477 try:478 while constructor.check_data():479 yield constructor.get_data()480 self.doc_infos.append(DocInfo(requested_version=version(self.version)))481 finally:482 parser.dispose()483 for comp in ('reader', 'scanner'):484 try:485 getattr(getattr(self, '_' + comp), f'reset_{comp}')()486 except AttributeError:487 pass488 489 def get_constructor_parser(self, stream: StreamTextType) -> Any:490 """491 the old cyaml needs special setup, and therefore the stream492 """493 if self.Constructor is None:494 if 'full' in self.typ:495 raise YAMLError(496 "\nyou can only use yaml=YAML(typ='full') for dumping\n", # NOQA497 )498 if self.Parser is not CParser:499 if self.Reader is None:500 self.Reader = ruamel.yaml.reader.Reader501 if self.Scanner is None:502 self.Scanner = ruamel.yaml.scanner.Scanner503 self.reader.stream = stream504 else:505 if self.Reader is not None:506 if self.Scanner is None:507 self.Scanner = ruamel.yaml.scanner.Scanner508 self.Parser = ruamel.yaml.parser.Parser509 self.reader.stream = stream510 elif self.Scanner is not None:511 if self.Reader is None:512 self.Reader = ruamel.yaml.reader.Reader513 self.Parser = ruamel.yaml.parser.Parser514 self.reader.stream = stream515 else:516 # combined C level reader>scanner>parser517 # does some calls to the resolver, e.g. BaseResolver.descend_resolver518 # if you just initialise the CParser, too much of resolver.py519 # is actually used520 rslvr = self.Resolver521 # if rslvr is ruamel.yaml.resolver.VersionedResolver:522 # rslvr = ruamel.yaml.resolver.Resolver523 524 class XLoader(self.Parser, self.Constructor, rslvr): # type: ignore525 def __init__(526 selfx,527 stream: StreamTextType,528 version: Optional[VersionType] = self.version,529 preserve_quotes: Optional[bool] = None,530 ) -> None:531 # NOQA532 CParser.__init__(selfx, stream)533 selfx._parser = selfx._composer = selfx534 self.Constructor.__init__(selfx, loader=selfx)535 selfx.allow_duplicate_keys = self.allow_duplicate_keys536 rslvr.__init__(selfx, version=version, loadumper=selfx)537 538 self._stream = stream539 loader = XLoader(stream)540 self._scanner = loader541 return loader, loader542 return self.constructor, self.parser543 544 def emit(self, events: Any, stream: Any) -> None:545 """546 Emit YAML parsing events into a stream.547 If stream is None, return the produced string instead.548 """549 _, _, emitter = self.get_serializer_representer_emitter(stream, None)550 try:551 for event in events:552 emitter.emit(event)553 finally:554 try:555 emitter.dispose()556 except AttributeError:557 raise558 559 def serialize(self, node: Any, stream: Optional[StreamType]) -> Any:560 """561 Serialize a representation tree into a YAML stream.562 If stream is None, return the produced string instead.563 """564 self.serialize_all([node], stream)565 566 def serialize_all(self, nodes: Any, stream: Optional[StreamType]) -> Any:567 """568 Serialize a sequence of representation trees into a YAML stream.569 If stream is None, return the produced string instead.570 """571 serializer, _, emitter = self.get_serializer_representer_emitter(stream, None)572 try:573 serializer.open()574 for node in nodes:575 serializer.serialize(node)576 serializer.close()577 finally:578 try:579 emitter.dispose()580 except AttributeError:581 raise582 583 def dump(584 self: Any, data: Union[Path, StreamType], stream: Any = None, *, transform: Any = None,585 ) -> Any:586 if self._context_manager:587 if not self._output:588 raise TypeError('Missing output stream while dumping from context manager')589 if transform is not None:590 x = self.__class__.__name__591 raise TypeError(592 f'{x}.dump() in the context manager cannot have transform keyword',593 )594 self._context_manager.dump(data)595 else: # old style596 if stream is None:597 raise TypeError('Need a stream argument when not dumping from context manager')598 return self.dump_all([data], stream, transform=transform)599 600 def dump_all(601 self, documents: Any, stream: Union[Path, StreamType], *, transform: Any = None,602 ) -> Any:603 if self._context_manager:604 raise NotImplementedError605 self._output = stream606 self._context_manager = YAMLContextManager(self, transform=transform)607 for data in documents:608 self._context_manager.dump(data)609 self._context_manager.teardown_output()610 self._output = None611 self._context_manager = None612 613 def Xdump_all(self, documents: Any, stream: Any, *, transform: Any = None) -> Any:614 """615 Serialize a sequence of Python objects into a YAML stream.616 """617 if not hasattr(stream, 'write') and hasattr(stream, 'open'):618 # pathlib.Path() instance619 with stream.open('w') as fp:620 return self.dump_all(documents, fp, transform=transform)621 # The stream should have the methods `write` and possibly `flush`.622 if self.top_level_colon_align is True:623 tlca: Any = max([len(str(x)) for x in documents[0]])624 else:625 tlca = self.top_level_colon_align626 if transform is not None:627 fstream = stream628 if self.encoding is None:629 stream = StringIO()630 else:631 stream = BytesIO()632 serializer, representer, emitter = self.get_serializer_representer_emitter(633 stream, tlca,634 )635 try:636 self.serializer.open()637 for data in documents:638 try:639 self.representer.represent(data)640 except AttributeError:641 # nprint(dir(dumper._representer))642 raise643 self.serializer.close()644 finally:645 try:646 self.emitter.dispose()647 except AttributeError:648 raise649 # self.dumper.dispose() # cyaml650 delattr(self, '_serializer') # NOQA651 delattr(self, '_emitter') # NOQA652 if transform:653 val = stream.getvalue()654 if self.encoding:655 val = val.decode(self.encoding)656 if fstream is None:657 transform(val)658 else:659 fstream.write(transform(val))660 return None661 662 def get_serializer_representer_emitter(self, stream: StreamType, tlca: Any) -> Any:663 # we have only .Serializer to deal with (vs .Reader & .Scanner), much simpler664 if self.Emitter is not CEmitter:665 if self.Serializer is None:666 self.Serializer = ruamel.yaml.serializer.Serializer667 self.emitter.stream = stream668 self.emitter.top_level_colon_align = tlca669 if self.scalar_after_indicator is not None:670 self.emitter.scalar_after_indicator = self.scalar_after_indicator671 return self.serializer, self.representer, self.emitter672 if self.Serializer is not None:673 # cannot set serializer with CEmitter674 self.Emitter = ruamel.yaml.emitter.Emitter675 self.emitter.stream = stream676 self.emitter.top_level_colon_align = tlca677 if self.scalar_after_indicator is not None:678 self.emitter.scalar_after_indicator = self.scalar_after_indicator679 return self.serializer, self.representer, self.emitter680 # C routines681 682 rslvr = (683 ruamel.yaml.resolver.BaseResolver684 if 'base' in self.typ685 else ruamel.yaml.resolver.Resolver686 )687 688 class XDumper(CEmitter, self.Representer, rslvr): # type: ignore689 def __init__(690 selfx: StreamType,691 stream: Any,692 default_style: Any = None,693 default_flow_style: Any = None,694 canonical: Optional[bool] = None,695 indent: Optional[int] = None,696 width: Optional[int] = None,697 allow_unicode: Optional[bool] = None,698 line_break: Any = None,699 encoding: Any = None,700 explicit_start: Optional[bool] = None,701 explicit_end: Optional[bool] = None,702 version: Any = None,703 tags: Any = None,704 block_seq_indent: Any = None,705 top_level_colon_align: Any = None,706 prefix_colon: Any = None,707 ) -> None:708 # NOQA709 CEmitter.__init__(710 selfx,711 stream,712 canonical=canonical,713 indent=indent,714 width=width,715 encoding=encoding,716 allow_unicode=allow_unicode,717 line_break=line_break,718 explicit_start=explicit_start,719 explicit_end=explicit_end,720 version=version,721 tags=tags,722 )723 selfx._emitter = selfx._serializer = selfx._representer = selfx724 self.Representer.__init__(725 selfx, default_style=default_style, default_flow_style=default_flow_style,726 )727 rslvr.__init__(selfx)728 729 self._stream = stream730 dumper = XDumper(731 stream,732 default_style=self.default_style,733 default_flow_style=self.default_flow_style,734 canonical=self.canonical,735 indent=self.old_indent,736 width=self.width,737 allow_unicode=self.allow_unicode,738 line_break=self.line_break,739 encoding=self.encoding,740 explicit_start=self.explicit_start,741 explicit_end=self.explicit_end,742 version=self.version,743 tags=self.tags,744 )745 self._emitter = self._serializer = dumper746 return dumper, dumper, dumper747 748 # basic types749 def map(self, **kw: Any) -> Any:750 if 'rt' in self.typ:751 return CommentedMap(**kw)752 else:753 return dict(**kw)754 755 def seq(self, *args: Any) -> Any:756 if 'rt' in self.typ:757 return CommentedSeq(*args)758 else:759 return list(*args)760 761 # helpers762 def official_plug_ins(self) -> Any:763 """search for list of subdirs that are plug-ins, if __file__ is not available, e.g.764 single file installers that are not properly emulating a file-system (issue 324)765 no plug-ins will be found. If any are packaged, you know which file that are766 and you can explicitly provide it during instantiation:767 yaml = ruamel.yaml.YAML(plug_ins=['ruamel/yaml/jinja2/__plug_in__'])768 """769 try:770 bd = os.path.dirname(__file__)771 except NameError:772 return []773 gpbd = os.path.dirname(os.path.dirname(bd))774 res = [x.replace(gpbd, "")[1:-3] for x in glob.glob(bd + '/*/__plug_in__.py')]775 return res776 777 def register_class(self, cls: Any) -> Any:778 """779 register a class for dumping/loading780 - if it has attribute yaml_tag use that to register, else use class name781 - if it has methods to_yaml/from_yaml use those to dump/load else dump attributes782 as mapping783 """784 tag = getattr(cls, 'yaml_tag', '!' + cls.__name__)785 try:786 self.representer.add_representer(cls, cls.to_yaml)787 except AttributeError:788 789 def t_y(representer: Any, data: Any) -> Any:790 return representer.represent_yaml_object(791 tag, data, cls, flow_style=representer.default_flow_style,792 )793 794 self.representer.add_representer(cls, t_y)795 try:796 self.constructor.add_constructor(tag, cls.from_yaml)797 except AttributeError:798 799 def f_y(constructor: Any, node: Any) -> Any:800 return constructor.construct_yaml_object(node, cls)801 802 self.constructor.add_constructor(tag, f_y)803 return cls804 805 # ### context manager806 807 def __enter__(self) -> Any:808 self._context_manager = YAMLContextManager(self)809 return self810 811 def __exit__(812 self,813 typ: Optional[Type[BaseException]],814 value: Optional[BaseException],815 traceback: Optional[TracebackType],816 ) -> None:817 if typ:818 nprint('typ', typ)819 self._context_manager.teardown_output()820 # self._context_manager.teardown_input()821 self._context_manager = None822 823 # ### backwards compatibility824 def _indent(self, mapping: Any = None, sequence: Any = None, offset: Any = None) -> None:825 if mapping is not None:826 self.map_indent = mapping827 if sequence is not None:828 self.sequence_indent = sequence829 if offset is not None:830 self.sequence_dash_offset = offset831 832 @property833 def version(self) -> Optional[Tuple[int, int]]:834 return self._version835 836 @version.setter837 def version(self, val: VersionType) -> None:838 if val is None:839 self._version = val840 return841 elif isinstance(val, str):842 sval = tuple(int(x) for x in val.split('.'))843 elif isinstance(val, (list, tuple)):844 sval = tuple(int(x) for x in val)845 elif isinstance(val, Version):846 sval = (val.major, val.minor)847 else:848 raise TypeError(f'unknown version type {type(val)}')849 assert len(sval) == 2, f'version can only have major.minor, got {val}'850 assert sval[0] == 1, f'version major part can only be 1, got {val}'851 assert sval[1] in [1, 2], f'version minor part can only be 2 or 1, got {val}'852 self._version = sval853 854 @property855 def tags(self) -> Any:856 return self._tags857 858 @tags.setter859 def tags(self, val: Any) -> None:860 self._tags = val861 862 @property863 def indent(self) -> Any:864 return self._indent865 866 @indent.setter867 def indent(self, val: Any) -> None:868 self.old_indent = val869 870 @property871 def block_seq_indent(self) -> Any:872 return self.sequence_dash_offset873 874 @block_seq_indent.setter875 def block_seq_indent(self, val: Any) -> None:876 self.sequence_dash_offset = val877 878 def compact(self, seq_seq: Any = None, seq_map: Any = None) -> None:879 self.compact_seq_seq = seq_seq880 self.compact_seq_map = seq_map881 882 883class YAMLContextManager:884 def __init__(self, yaml: Any, transform: Any = None) -> None:885 # used to be: (Any, Optional[Callable]) -> None886 self._yaml = yaml887 self._output_inited = False888 self._output_path = None889 self._output = self._yaml._output890 self._transform = transform891 892 # self._input_inited = False893 # self._input = input894 # self._input_path = None895 # self._transform = yaml.transform896 # self._fstream = None897 898 if not hasattr(self._output, 'write') and hasattr(self._output, 'open'):899 # pathlib.Path() instance, open with the same mode900 self._output_path = self._output901 self._output = self._output_path.open('w')902 903 # if not hasattr(self._stream, 'write') and hasattr(stream, 'open'):904 # if not hasattr(self._input, 'read') and hasattr(self._input, 'open'):905 # # pathlib.Path() instance, open with the same mode906 # self._input_path = self._input907 # self._input = self._input_path.open('r')908 909 if self._transform is not None:910 self._fstream = self._output911 if self._yaml.encoding is None:912 self._output = StringIO()913 else:914 self._output = BytesIO()915 916 def teardown_output(self) -> None:917 if self._output_inited:918 self._yaml.serializer.close()919 else:920 return921 try:922 self._yaml.emitter.dispose()923 except AttributeError:924 raise925 # self.dumper.dispose() # cyaml926 try:927 delattr(self._yaml, '_serializer') # NOQA928 delattr(self._yaml, '_emitter') # NOQA929 except AttributeError:930 raise931 if self._transform:932 val = self._output.getvalue()933 if self._yaml.encoding:934 val = val.decode(self._yaml.encoding)935 if self._fstream is None:936 self._transform(val)937 else:938 self._fstream.write(self._transform(val))939 self._fstream.flush()940 self._output = self._fstream # maybe not necessary941 if self._output_path is not None:942 self._output.close()943 944 def init_output(self, first_data: Any) -> None:945 if self._yaml.top_level_colon_align is True:946 tlca: Any = max([len(str(x)) for x in first_data])947 else:948 tlca = self._yaml.top_level_colon_align949 self._yaml.get_serializer_representer_emitter(self._output, tlca)950 self._yaml.serializer.open()951 self._output_inited = True952 953 def dump(self, data: Any) -> None:954 if not self._output_inited:955 self.init_output(data)956 try:957 self._yaml.representer.represent(data)958 except AttributeError:959 # nprint(dir(dumper._representer))960 raise961 962 # def teardown_input(self):963 # pass964 #965 # def init_input(self):966 # # set the constructor and parser on YAML() instance967 # self._yaml.get_constructor_parser(stream)968 #969 # def load(self):970 # if not self._input_inited:971 # self.init_input()972 # try:973 # while self._yaml.constructor.check_data():974 # yield self._yaml.constructor.get_data()975 # finally:976 # parser.dispose()977 # try:978 # self._reader.reset_reader() # type: ignore979 # except AttributeError:980 # pass981 # try:982 # self._scanner.reset_scanner() # type: ignore983 # except AttributeError:984 # pass985 986 987def yaml_object(yml: Any) -> Any:988 """ decorator for classes that needs to dump/load objects989 The tag for such objects is taken from the class attribute yaml_tag (or the990 class name in lowercase in case unavailable)991 If methods to_yaml and/or from_yaml are available, these are called for dumping resp.992 loading, default routines (dumping a mapping of the attributes) used otherwise.993 """994 995 def yo_deco(cls: Any) -> Any:996 tag = getattr(cls, 'yaml_tag', '!' + cls.__name__)997 try:998 yml.representer.add_representer(cls, cls.to_yaml)999 except AttributeError:1000 1001 def t_y(representer: Any, data: Any) -> Any:1002 return representer.represent_yaml_object(1003 tag, data, cls, flow_style=representer.default_flow_style,1004 )1005 1006 yml.representer.add_representer(cls, t_y)1007 try:1008 yml.constructor.add_constructor(tag, cls.from_yaml)1009 except AttributeError:1010 1011 def f_y(constructor: Any, node: Any) -> Any:1012 return constructor.construct_yaml_object(node, cls)1013 1014 yml.constructor.add_constructor(tag, f_y)1015 return cls1016 1017 return yo_deco1018 1019 1020########################################################################################1021def warn_deprecation(fun: Any, method: Any, arg: str = '') -> None:1022 warnings.warn(1023 f'\n{fun} will be removed, use\n\n yaml=YAML({arg})\n yaml.{method}(...)\n\ninstead', # NOQA1024 PendingDeprecationWarning, # this will show when testing with pytest/tox1025 stacklevel=3,1026 )1027 1028 1029def error_deprecation(fun: Any, method: Any, arg: str = '', comment: str = 'instead of') -> None: # NOQA1030 import inspect1031 1032 s = f'\n"{fun}()" has been removed, use\n\n yaml = YAML({arg})\n yaml.{method}(...)\n\n{comment}' # NOQA1033 try:1034 info = inspect.getframeinfo(inspect.stack()[2][0])1035 context = '' if info.code_context is None else "".join(info.code_context)1036 s += f' file "{info.filename}", line {info.lineno}\n\n{context}'1037 except Exception as e:1038 _ = e1039 s += '\n'1040 if sys.version_info < (3, 10):1041 raise AttributeError(s)1042 else:1043 raise AttributeError(s, name=None)1044 1045 1046_error_dep_arg = "typ='rt'"1047_error_dep_comment = "and register any classes that you use, or check the tag attribute on the loaded data,\ninstead of" # NOQA1048 1049########################################################################################1050 1051 1052def scan(stream: StreamTextType, Loader: Any = Loader) -> Any:1053 """1054 Scan a YAML stream and produce scanning tokens.1055 """1056 error_deprecation('scan', 'scan', arg=_error_dep_arg, comment=_error_dep_comment)1057 1058 1059def parse(stream: StreamTextType, Loader: Any = Loader) -> Any:1060 """1061 Parse a YAML stream and produce parsing events.1062 """1063 error_deprecation('parse', 'parse', arg=_error_dep_arg, comment=_error_dep_comment)1064 1065 1066def compose(stream: StreamTextType, Loader: Any = Loader) -> Any:1067 """1068 Parse the first YAML document in a stream1069 and produce the corresponding representation tree.1070 """1071 error_deprecation('compose', 'compose', arg=_error_dep_arg, comment=_error_dep_comment)1072 1073 1074def compose_all(stream: StreamTextType, Loader: Any = Loader) -> Any:1075 """1076 Parse all YAML documents in a stream1077 and produce corresponding representation trees.1078 """1079 error_deprecation('compose', 'compose', arg=_error_dep_arg, comment=_error_dep_comment)1080 1081 1082def load(1083 stream: Any, Loader: Any = None, version: Any = None, preserve_quotes: Any = None,1084) -> Any:1085 """1086 Parse the first YAML document in a stream1087 and produce the corresponding Python object.1088 """1089 error_deprecation('load', 'load', arg=_error_dep_arg, comment=_error_dep_comment)1090 1091 1092def load_all(1093 stream: Any, Loader: Any = None, version: Any = None, preserve_quotes: Any = None,1094) -> Any:1095 # NOQA1096 """1097 Parse all YAML documents in a stream1098 and produce corresponding Python objects.1099 """1100 error_deprecation('load_all', 'load_all', arg=_error_dep_arg, comment=_error_dep_comment)1101 1102 1103def safe_load(stream: StreamTextType, version: Optional[VersionType] = None) -> Any:1104 """1105 Parse the first YAML document in a stream1106 and produce the corresponding Python object.1107 Resolve only basic YAML tags.1108 """1109 error_deprecation('safe_load', 'load', arg="typ='safe', pure=True")1110 1111 1112def safe_load_all(stream: StreamTextType, version: Optional[VersionType] = None) -> Any:1113 """1114 Parse all YAML documents in a stream1115 and produce corresponding Python objects.1116 Resolve only basic YAML tags.1117 """1118 error_deprecation('safe_load_all', 'load_all', arg="typ='safe', pure=True")1119 1120 1121def round_trip_load(1122 stream: StreamTextType,1123 version: Optional[VersionType] = None,1124 preserve_quotes: Optional[bool] = None,1125) -> Any:1126 """1127 Parse the first YAML document in a stream1128 and produce the corresponding Python object.1129 Resolve only basic YAML tags.1130 """1131 error_deprecation('round_trip_load_all', 'load')1132 1133 1134def round_trip_load_all(1135 stream: StreamTextType,1136 version: Optional[VersionType] = None,1137 preserve_quotes: Optional[bool] = None,1138) -> Any:1139 """1140 Parse all YAML documents in a stream1141 and produce corresponding Python objects.1142 Resolve only basic YAML tags.1143 """1144 error_deprecation('round_trip_load_all', 'load_all')1145 1146 1147def emit(1148 events: Any,1149 stream: Optional[StreamType] = None,1150 Dumper: Any = Dumper,1151 canonical: Optional[bool] = None,1152 indent: Union[int, None] = None,1153 width: Optional[int] = None,1154 allow_unicode: Optional[bool] = None,1155 line_break: Any = None,1156) -> Any:1157 # NOQA1158 """1159 Emit YAML parsing events into a stream.1160 If stream is None, return the produced string instead.1161 """1162 error_deprecation('emit', 'emit', arg="typ='safe', pure=True")1163 1164 1165enc = None1166 1167 1168def serialize_all(1169 nodes: Any,1170 stream: Optional[StreamType] = None,1171 Dumper: Any = Dumper,1172 canonical: Any = None,1173 indent: Optional[int] = None,1174 width: Optional[int] = None,1175 allow_unicode: Optional[bool] = None,1176 line_break: Any = None,1177 encoding: Any = enc,1178 explicit_start: Optional[bool] = None,1179 explicit_end: Optional[bool] = None,1180 version: Optional[VersionType] = None,1181 tags: Any = None,1182) -> Any:1183 # NOQA1184 """1185 Serialize a sequence of representation trees into a YAML stream.1186 If stream is None, return the produced string instead.1187 """1188 error_deprecation('serialize_all', 'serialize_all', arg="typ='safe', pure=True")1189 1190 1191def serialize(1192 node: Any, stream: Optional[StreamType] = None, Dumper: Any = Dumper, **kwds: Any,1193) -> Any:1194 """1195 Serialize a representation tree into a YAML stream.1196 If stream is None, return the produced string instead.1197 """1198 error_deprecation('serialize', 'serialize', arg="typ='safe', pure=True")1199 1200