codekingpro/portable-devtools

1 2from __future__ import annotations3 4import re5 6if False:  # MYPY7    from typing import Any, Dict, List, Union, Text, Optional  # NOQA8    from ruamel.yaml.compat import VersionType  # NOQA9 10from ruamel.yaml.tag import Tag11from ruamel.yaml.compat import _DEFAULT_YAML_VERSION  # NOQA12from ruamel.yaml.error import *  # NOQA13from ruamel.yaml.nodes import MappingNode, ScalarNode, SequenceNode  # NOQA14from ruamel.yaml.util import RegExp  # NOQA15 16__all__ = ['BaseResolver', 'Resolver', 'VersionedResolver']17 18 19# fmt: off20# resolvers consist of21# - a list of applicable version22# - a tag23# - a regexp24# - a list of first characters to match25implicit_resolvers = [26    ([(1, 2)],27        'tag:yaml.org,2002:bool',28        RegExp('''^(?:true|True|TRUE|false|False|FALSE)$''', re.X),29        list('tTfF')),30    ([(1, 1)],31        'tag:yaml.org,2002:bool',32        RegExp('''^(?:y|Y|yes|Yes|YES|n|N|no|No|NO33        |true|True|TRUE|false|False|FALSE34        |on|On|ON|off|Off|OFF)$''', re.X),35        list('yYnNtTfFoO')),36    ([(1, 2)],37        'tag:yaml.org,2002:float',38        RegExp('''^(?:39         [-+]?(?:[0-9][0-9_]*)\\.[0-9_]*(?:[eE][-+]?[0-9]+)?40        |[-+]?(?:[0-9][0-9_]*)(?:[eE][-+]?[0-9]+)41        |[-+]?\\.[0-9_]+(?:[eE][-+][0-9]+)?42        |[-+]?\\.(?:inf|Inf|INF)43        |\\.(?:nan|NaN|NAN))$''', re.X),44        list('-+0123456789.')),45    ([(1, 1)],46        'tag:yaml.org,2002:float',47        RegExp('''^(?:48         [-+]?(?:[0-9][0-9_]*)\\.[0-9_]*(?:[eE][-+]?[0-9]+)?49        |[-+]?(?:[0-9][0-9_]*)(?:[eE][-+]?[0-9]+)50        |\\.[0-9_]+(?:[eE][-+][0-9]+)?51        |[-+]?[0-9][0-9_]*(?::[0-5]?[0-9])+\\.[0-9_]*  # sexagesimal float52        |[-+]?\\.(?:inf|Inf|INF)53        |\\.(?:nan|NaN|NAN))$''', re.X),54        list('-+0123456789.')),55    ([(1, 2)],56        'tag:yaml.org,2002:int',57        RegExp('''^(?:[-+]?0b[0-1_]+58        |[-+]?0o?[0-7_]+59        |[-+]?[0-9_]+60        |[-+]?0x[0-9a-fA-F_]+)$''', re.X),61        list('-+0123456789')),62    ([(1, 1)],63        'tag:yaml.org,2002:int',64        RegExp('''^(?:[-+]?0b[0-1_]+65        |[-+]?0?[0-7_]+66        |[-+]?(?:0|[1-9][0-9_]*)67        |[-+]?0x[0-9a-fA-F_]+68        |[-+]?[1-9][0-9_]*(?::[0-5]?[0-9])+)$''', re.X),  # sexagesimal int69        list('-+0123456789')),70    ([(1, 2), (1, 1)],71        'tag:yaml.org,2002:merge',72        RegExp('^(?:<<)$'),73        ['<']),74    ([(1, 2), (1, 1)],75        'tag:yaml.org,2002:null',76        RegExp('''^(?: ~77        |null|Null|NULL78        | )$''', re.X),79        ['~', 'n', 'N', '']),80    ([(1, 2), (1, 1)],81        'tag:yaml.org,2002:timestamp',82        RegExp('''^(?:[0-9][0-9][0-9][0-9]-[0-9][0-9]-[0-9][0-9]83        |[0-9][0-9][0-9][0-9] -[0-9][0-9]? -[0-9][0-9]?84        (?:[Tt]|[ \\t]+)[0-9][0-9]?85        :[0-9][0-9] :[0-9][0-9] (?:\\.[0-9]*)?86        (?:[ \\t]*(?:Z|[-+][0-9][0-9]?(?::[0-9][0-9])?))?)$''', re.X),87        list('0123456789')),88    ([(1, 2), (1, 1)],89        'tag:yaml.org,2002:value',90        RegExp('^(?:=)$'),91        ['=']),92    # The following resolver is only for documentation purposes. It cannot work93    # because plain scalars cannot start with '!', '&', or '*'.94    ([(1, 2), (1, 1)],95        'tag:yaml.org,2002:yaml',96        RegExp('^(?:!|&|\\*)$'),97        list('!&*')),98]99# fmt: on100 101 102class ResolverError(YAMLError):103    pass104 105 106class BaseResolver:107 108    DEFAULT_SCALAR_TAG = Tag(suffix='tag:yaml.org,2002:str')109    DEFAULT_SEQUENCE_TAG = Tag(suffix='tag:yaml.org,2002:seq')110    DEFAULT_MAPPING_TAG = Tag(suffix='tag:yaml.org,2002:map')111 112    yaml_implicit_resolvers: Dict[Any, Any] = {}113    yaml_path_resolvers: Dict[Any, Any] = {}114 115    def __init__(self: Any, loadumper: Any = None) -> None:116        self.loadumper = loadumper117        if self.loadumper is not None and getattr(self.loadumper, '_resolver', None) is None:118            self.loadumper._resolver = self.loadumper119        self._loader_version: Any = None120        self.resolver_exact_paths: List[Any] = []121        self.resolver_prefix_paths: List[Any] = []122 123    @property124    def parser(self) -> Any:125        if self.loadumper is not None:126            if hasattr(self.loadumper, 'typ'):127                return self.loadumper.parser128            return self.loadumper._parser129        return None130 131    @classmethod132    def add_implicit_resolver_base(cls, tag: Any, regexp: Any, first: Any) -> None:133        if 'yaml_implicit_resolvers' not in cls.__dict__:134            # deepcopy doesn't work here135            cls.yaml_implicit_resolvers = {136                k: cls.yaml_implicit_resolvers[k][:] for k in cls.yaml_implicit_resolvers137            }138        if first is None:139            first = [None]140        for ch in first:141            cls.yaml_implicit_resolvers.setdefault(ch, []).append((tag, regexp))142 143    @classmethod144    def add_implicit_resolver(cls, tag: Any, regexp: Any, first: Any) -> None:145        if 'yaml_implicit_resolvers' not in cls.__dict__:146            # deepcopy doesn't work here147            cls.yaml_implicit_resolvers = {148                k: cls.yaml_implicit_resolvers[k][:] for k in cls.yaml_implicit_resolvers149            }150        if first is None:151            first = [None]152        for ch in first:153            cls.yaml_implicit_resolvers.setdefault(ch, []).append((tag, regexp))154        implicit_resolvers.append(([(1, 2), (1, 1)], tag, regexp, first))155 156    # @classmethod157    # def add_implicit_resolver(cls, tag, regexp, first):158 159    @classmethod160    def add_path_resolver(cls, tag: Any, path: Any, kind: Any = None) -> None:161        # Note: `add_path_resolver` is experimental.  The API could be changed.162        # `new_path` is a pattern that is matched against the path from the163        # root to the node that is being considered.  `node_path` elements are164        # tuples `(node_check, index_check)`.  `node_check` is a node class:165        # `ScalarNode`, `SequenceNode`, `MappingNode` or `None`.  `None`166        # matches any kind of a node.  `index_check` could be `None`, a boolean167        # value, a string value, or a number.  `None` and `False` match against168        # any _value_ of sequence and mapping nodes.  `True` matches against169        # any _key_ of a mapping node.  A string `index_check` matches against170        # a mapping value that corresponds to a scalar key which content is171        # equal to the `index_check` value.  An integer `index_check` matches172        # against a sequence value with the index equal to `index_check`.173        if 'yaml_path_resolvers' not in cls.__dict__:174            cls.yaml_path_resolvers = cls.yaml_path_resolvers.copy()175        new_path: List[Any] = []176        for element in path:177            if isinstance(element, (list, tuple)):178                if len(element) == 2:179                    node_check, index_check = element180                elif len(element) == 1:181                    node_check = element[0]182                    index_check = True183                else:184                    raise ResolverError(f'Invalid path element: {element!s}')185            else:186                node_check = None187                index_check = element188            if node_check is str:189                node_check = ScalarNode190            elif node_check is list:191                node_check = SequenceNode192            elif node_check is dict:193                node_check = MappingNode194            elif (195                node_check not in [ScalarNode, SequenceNode, MappingNode]196                and not isinstance(node_check, str)197                and node_check is not None198            ):199                raise ResolverError(f'Invalid node checker: {node_check!s}')200            if not isinstance(index_check, (str, int)) and index_check is not None:201                raise ResolverError(f'Invalid index checker: {index_check!s}')202            new_path.append((node_check, index_check))203        if kind is str:204            kind = ScalarNode205        elif kind is list:206            kind = SequenceNode207        elif kind is dict:208            kind = MappingNode209        elif kind not in [ScalarNode, SequenceNode, MappingNode] and kind is not None:210            raise ResolverError(f'Invalid node kind: {kind!s}')211        cls.yaml_path_resolvers[tuple(new_path), kind] = tag212 213    def descend_resolver(self, current_node: Any, current_index: Any) -> None:214        if not self.yaml_path_resolvers:215            return216        exact_paths = {}217        prefix_paths = []218        if current_node:219            depth = len(self.resolver_prefix_paths)220            for path, kind in self.resolver_prefix_paths[-1]:221                if self.check_resolver_prefix(depth, path, kind, current_node, current_index):222                    if len(path) > depth:223                        prefix_paths.append((path, kind))224                    else:225                        exact_paths[kind] = self.yaml_path_resolvers[path, kind]226        else:227            for path, kind in self.yaml_path_resolvers:228                if not path:229                    exact_paths[kind] = self.yaml_path_resolvers[path, kind]230                else:231                    prefix_paths.append((path, kind))232        self.resolver_exact_paths.append(exact_paths)233        self.resolver_prefix_paths.append(prefix_paths)234 235    def ascend_resolver(self) -> None:236        if not self.yaml_path_resolvers:237            return238        self.resolver_exact_paths.pop()239        self.resolver_prefix_paths.pop()240 241    def check_resolver_prefix(242        self, depth: int, path: Any, kind: Any, current_node: Any, current_index: Any,243    ) -> bool:244        node_check, index_check = path[depth - 1]245        if isinstance(node_check, str):246            if current_node.tag != node_check:247                return False248        elif node_check is not None:249            if not isinstance(current_node, node_check):250                return False251        if index_check is True and current_index is not None:252            return False253        if (index_check is False or index_check is None) and current_index is None:254            return False255        if isinstance(index_check, str):256            if not (257                isinstance(current_index, ScalarNode) and index_check == current_index.value258            ):259                return False260        elif isinstance(index_check, int) and not isinstance(index_check, bool):261            if index_check != current_index:262                return False263        return True264 265    def resolve(self, kind: Any, value: Any, implicit: Any) -> Any:266        if kind is ScalarNode and implicit[0]:267            if value == "":268                resolvers = self.yaml_implicit_resolvers.get("", [])269            else:270                resolvers = self.yaml_implicit_resolvers.get(value[0], [])271            resolvers += self.yaml_implicit_resolvers.get(None, [])272            for tag, regexp in resolvers:273                if regexp.match(value):274                    return Tag(suffix=tag)275            implicit = implicit[1]276        if bool(self.yaml_path_resolvers):277            exact_paths = self.resolver_exact_paths[-1]278            if kind in exact_paths:279                return Tag(suffix=exact_paths[kind])280            if None in exact_paths:281                return Tag(suffix=exact_paths[None])282        if kind is ScalarNode:283            return self.DEFAULT_SCALAR_TAG284        elif kind is SequenceNode:285            return self.DEFAULT_SEQUENCE_TAG286        elif kind is MappingNode:287            return self.DEFAULT_MAPPING_TAG288 289    @property290    def processing_version(self) -> Any:291        return None292 293 294class Resolver(BaseResolver):295    pass296 297 298for ir in implicit_resolvers:299    if (1, 2) in ir[0]:300        Resolver.add_implicit_resolver_base(*ir[1:])301 302 303class VersionedResolver(BaseResolver):304    """305    contrary to the "normal" resolver, the smart resolver delays loading306    the pattern matching rules. That way it can decide to load 1.1 rules307    or the (default) 1.2 rules, that no longer support octal without 0o, sexagesimals308    and Yes/No/On/Off booleans.309    """310 311    def __init__(312        self, version: Optional[VersionType] = None, loader: Any = None, loadumper: Any = None,313    ) -> None:314        if loader is None and loadumper is not None:315            loader = loadumper316        BaseResolver.__init__(self, loader)317        self._loader_version = self.get_loader_version(version)318        self._version_implicit_resolver: Dict[Any, Any] = {}319 320    def add_version_implicit_resolver(321        self, version: VersionType, tag: Any, regexp: Any, first: Any,322    ) -> None:323        if first is None:324            first = [None]325        impl_resolver = self._version_implicit_resolver.setdefault(version, {})326        for ch in first:327            impl_resolver.setdefault(ch, []).append((tag, regexp))328 329    def get_loader_version(self, version: Optional[VersionType]) -> Any:330        if version is None or isinstance(version, tuple):331            return version332        if isinstance(version, list):333            return tuple(version)334        # assume string335        assert isinstance(version, str)336        return tuple(map(int, version.split('.')))337 338    @property339    def versioned_resolver(self) -> Any:340        """341        select the resolver based on the version we are parsing342        """343        version = self.processing_version344        if isinstance(version, str):345            version = tuple(map(int, version.split('.')))346        if version not in self._version_implicit_resolver:347            for x in implicit_resolvers:348                if version in x[0]:349                    self.add_version_implicit_resolver(version, x[1], x[2], x[3])350        return self._version_implicit_resolver[version]351 352    def resolve(self, kind: Any, value: Any, implicit: Any) -> Any:353        if kind is ScalarNode and implicit[0]:354            if value == "":355                resolvers = self.versioned_resolver.get("", [])356            else:357                resolvers = self.versioned_resolver.get(value[0], [])358            resolvers += self.versioned_resolver.get(None, [])359            for tag, regexp in resolvers:360                if regexp.match(value):361                    return Tag(suffix=tag)362            implicit = implicit[1]363        if bool(self.yaml_path_resolvers):364            exact_paths = self.resolver_exact_paths[-1]365            if kind in exact_paths:366                return Tag(suffix=exact_paths[kind])367            if None in exact_paths:368                return Tag(suffix=exact_paths[None])369        if kind is ScalarNode:370            return self.DEFAULT_SCALAR_TAG371        elif kind is SequenceNode:372            return self.DEFAULT_SEQUENCE_TAG373        elif kind is MappingNode:374            return self.DEFAULT_MAPPING_TAG375 376    @property377    def processing_version(self) -> Any:378        try:379            version = self.loadumper._scanner.yaml_version380        except AttributeError:381            try:382                if hasattr(self.loadumper, 'typ'):383                    version = self.loadumper.version384                else:385                    version = self.loadumper._serializer.use_version  # dumping386            except AttributeError:387                version = None388        if version is None:389            version = self._loader_version390            if version is None:391                version = _DEFAULT_YAML_VERSION392        return version393