codekingpro/portable-devtools

1 2from __future__ import annotations3 4import warnings5 6from ruamel.yaml.error import MarkedYAMLError, ReusedAnchorWarning7from ruamel.yaml.compat import nprint, nprintf  # NOQA8 9from ruamel.yaml.events import (10    StreamStartEvent,11    StreamEndEvent,12    MappingStartEvent,13    MappingEndEvent,14    SequenceStartEvent,15    SequenceEndEvent,16    AliasEvent,17    ScalarEvent,18)19from ruamel.yaml.nodes import MappingNode, ScalarNode, SequenceNode20 21if False:  # MYPY22    from typing import Any, Dict, Optional, List  # NOQA23 24__all__ = ['Composer', 'ComposerError', 'MaxDepthExceededError']25 26 27class ComposerError(MarkedYAMLError):28    pass29 30 31class MaxDepthExceededError(MarkedYAMLError):32    pass33 34 35class Composer:36    def __init__(self, loader: Any = None) -> None:37        self.loader = loader38        if self.loader is not None and getattr(self.loader, '_composer', None) is None:39            self.loader._composer = self40        self.anchors: Dict[Any, Any] = {}41        self.warn_double_anchors = True42        self.depth = 043 44    @property45    def parser(self) -> Any:46        if hasattr(self.loader, 'typ'):47            self.loader.parser48        return self.loader._parser49 50    @property51    def resolver(self) -> Any:52        # assert self.loader._resolver is not None53        if hasattr(self.loader, 'typ'):54            self.loader.resolver55        return self.loader._resolver56 57    def check_node(self) -> Any:58        # Drop the STREAM-START event.59        if self.parser.check_event(StreamStartEvent):60            self.parser.get_event()61 62        # If there are more documents available?63        return not self.parser.check_event(StreamEndEvent)64 65    def get_node(self) -> Any:66        # Get the root node of the next document.67        if not self.parser.check_event(StreamEndEvent):68            return self.compose_document()69 70    def get_single_node(self) -> Any:71        # Drop the STREAM-START event.72        self.parser.get_event()73 74        # Compose a document if the stream is not empty.75        document: Any = None76        if not self.parser.check_event(StreamEndEvent):77            document = self.compose_document()78 79        # Ensure that the stream contains no more documents.80        if not self.parser.check_event(StreamEndEvent):81            event = self.parser.get_event()82            raise ComposerError(83                'expected a single document in the stream',84                document.start_mark,85                'but found another document',86                event.start_mark,87            )88 89        # Drop the STREAM-END event.90        self.parser.get_event()91 92        return document93 94    def compose_document(self: Any) -> Any:95        self.anchors = {}96        # Drop the DOCUMENT-START event.97        self.parser.get_event()98 99        # Compose the root node.100        node = self.compose_node(None, None)101 102        # Drop the DOCUMENT-END event.103        self.parser.get_event()104 105        return node106 107    def return_alias(self, a: Any) -> Any:108        return a109 110    def compose_node(self, parent: Any, index: Any) -> Any:111        if self.parser.check_event(AliasEvent):112            event = self.parser.get_event()113            alias = event.anchor114            if alias not in self.anchors:115                raise ComposerError(116                    None, None, f'found undefined alias {alias!r}', event.start_mark,117                )118            return self.return_alias(self.anchors[alias])119        self.depth += 1120        event = self.parser.peek_event()121        if self.loader.max_depth and self.depth > self.loader.max_depth:122            raise MaxDepthExceededError(123                None,124                None,125                f'maximum depth of data structure exceeded ({self.depth}), '126                'if necessary increase YAML().max_depth',127                event.start_mark,128            )129        anchor = event.anchor130        if anchor is not None:  # have an anchor131            if self.warn_double_anchors and anchor in self.anchors:132                ws = (133                    f'\nfound duplicate anchor {anchor!r}\n'134                    f'first occurrence {self.anchors[anchor].start_mark}\n'135                    f'second occurrence {event.start_mark}'136                )137                warnings.warn(ws, ReusedAnchorWarning, stacklevel=2)138        self.resolver.descend_resolver(parent, index)139        if self.parser.check_event(ScalarEvent):140            node = self.compose_scalar_node(anchor)141        elif self.parser.check_event(SequenceStartEvent):142            node = self.compose_sequence_node(anchor)143        elif self.parser.check_event(MappingStartEvent):144            node = self.compose_mapping_node(anchor)145        self.resolver.ascend_resolver()146        self.depth -= 1147        return node148 149    def compose_scalar_node(self, anchor: Any) -> Any:150        event = self.parser.get_event()151        tag = event.ctag152        if tag is None or str(tag) == '!':153            tag = self.resolver.resolve(ScalarNode, event.value, event.implicit)154            assert not isinstance(tag, str)155            # e.g tag.yaml.org,2002:str156        node = ScalarNode(157            tag,158            event.value,159            event.start_mark,160            event.end_mark,161            style=event.style,162            comment=event.comment,163            anchor=anchor,164        )165        if anchor is not None:166            self.anchors[anchor] = node167        return node168 169    def compose_sequence_node(self, anchor: Any) -> Any:170        start_event = self.parser.get_event()171        tag = start_event.ctag172        if tag is None or str(tag) == '!':173            tag = self.resolver.resolve(SequenceNode, None, start_event.implicit)174            assert not isinstance(tag, str)175        node = SequenceNode(176            tag,177            [],178            start_event.start_mark,179            None,180            flow_style=start_event.flow_style,181            comment=start_event.comment,182            anchor=anchor,183        )184        if anchor is not None:185            self.anchors[anchor] = node186        index = 0187        while not self.parser.check_event(SequenceEndEvent):188            node.value.append(self.compose_node(node, index))189            index += 1190        end_event = self.parser.get_event()191        if node.flow_style is True and end_event.comment is not None:192            if node.comment is not None:193                x = node.flow_style194                nprint(195                    f'Warning: unexpected end_event commment in sequence node {x}\n',196                    '    if possible, please report an issue with reproducable data/code',197                )198            node.comment = end_event.comment199        node.end_mark = end_event.end_mark200        self.check_end_doc_comment(end_event, node)201        return node202 203    def compose_mapping_node(self, anchor: Any) -> Any:204        start_event = self.parser.get_event()205        tag = start_event.ctag206        if tag is None or str(tag) == '!':207            tag = self.resolver.resolve(MappingNode, None, start_event.implicit)208            assert not isinstance(tag, str)209        node = MappingNode(210            tag,211            [],212            start_event.start_mark,213            None,214            flow_style=start_event.flow_style,215            comment=start_event.comment,216            anchor=anchor,217        )218        if anchor is not None:219            self.anchors[anchor] = node220        while not self.parser.check_event(MappingEndEvent):221            # key_event = self.parser.peek_event()222            item_key = self.compose_node(node, None)223            # if item_key in node.value:224            #     raise ComposerError("while composing a mapping",225            #             start_event.start_mark,226            #             "found duplicate key", key_event.start_mark)227            item_value = self.compose_node(node, item_key)228            # node.value[item_key] = item_value229            node.value.append((item_key, item_value))230        end_event = self.parser.get_event()231        if node.flow_style is True and end_event.comment is not None:232            node.comment = end_event.comment233        node.end_mark = end_event.end_mark234        self.check_end_doc_comment(end_event, node)235        return node236 237    def check_end_doc_comment(self, end_event: Any, node: Any) -> None:238        if end_event.comment and end_event.comment[1]:239            # pre comments on an end_event, no following to move to240            if node.comment is None:241                node.comment = [None, None]242            assert not isinstance(node, ScalarEvent)243            # this is a post comment on a mapping node, add as third element244            # in the list245            node.comment.append(end_event.comment[1])246            end_event.comment[1] = None247