git.reasoningtechnology.com Git - SubU/blob

   1 # This file is dual licensed under the terms of the Apache License, Version
   2 # 2.0, and the BSD License. See the LICENSE file in the root of this repository
   3 # for complete details.
   4
   5 import operator
   6 import os
   7 import platform
   8 import sys
   9 from typing import Any, Callable, Dict, List, Optional, Tuple, Union
  10
  11 from pkg_resources.extern.pyparsing import (  # noqa: N817
  12     Forward,
  13     Group,
  14     Literal as L,
  15     ParseException,
  16     ParseResults,
  17     QuotedString,
  18     ZeroOrMore,
  19     stringEnd,
  20     stringStart,
  21 )
  22
  23 from .specifiers import InvalidSpecifier, Specifier
  24
  25 __all__ = [
  26     "InvalidMarker",
  27     "UndefinedComparison",
  28     "UndefinedEnvironmentName",
  29     "Marker",
  30     "default_environment",
  31 ]
  32
  33 Operator = Callable[[str, str], bool]
  34
  35
  36 class InvalidMarker(ValueError):
  37     """
  38     An invalid marker was found, users should refer to PEP 508.
  39     """
  40
  41
  42 class UndefinedComparison(ValueError):
  43     """
  44     An invalid operation was attempted on a value that doesn't support it.
  45     """
  46
  47
  48 class UndefinedEnvironmentName(ValueError):
  49     """
  50     A name was attempted to be used that does not exist inside of the
  51     environment.
  52     """
  53
  54
  55 class Node:
  56     def __init__(self, value: Any) -> None:
  57         self.value = value
  58
  59     def __str__(self) -> str:
  60         return str(self.value)
  61
  62     def __repr__(self) -> str:
  63         return f"<{self.__class__.__name__}('{self}')>"
  64
  65     def serialize(self) -> str:
  66         raise NotImplementedError
  67
  68
  69 class Variable(Node):
  70     def serialize(self) -> str:
  71         return str(self)
  72
  73
  74 class Value(Node):
  75     def serialize(self) -> str:
  76         return f'"{self}"'
  77
  78
  79 class Op(Node):
  80     def serialize(self) -> str:
  81         return str(self)
  82
  83
  84 VARIABLE = (
  85     L("implementation_version")
  86     | L("platform_python_implementation")
  87     | L("implementation_name")
  88     | L("python_full_version")
  89     | L("platform_release")
  90     | L("platform_version")
  91     | L("platform_machine")
  92     | L("platform_system")
  93     | L("python_version")
  94     | L("sys_platform")
  95     | L("os_name")
  96     | L("os.name")  # PEP-345
  97     | L("sys.platform")  # PEP-345
  98     | L("platform.version")  # PEP-345
  99     | L("platform.machine")  # PEP-345
 100     | L("platform.python_implementation")  # PEP-345
 101     | L("python_implementation")  # undocumented setuptools legacy
 102     | L("extra")  # PEP-508
 103 )
 104 ALIASES = {
 105     "os.name": "os_name",
 106     "sys.platform": "sys_platform",
 107     "platform.version": "platform_version",
 108     "platform.machine": "platform_machine",
 109     "platform.python_implementation": "platform_python_implementation",
 110     "python_implementation": "platform_python_implementation",
 111 }
 112 VARIABLE.setParseAction(lambda s, l, t: Variable(ALIASES.get(t[0], t[0])))
 113
 114 VERSION_CMP = (
 115     L("===") | L("==") | L(">=") | L("<=") | L("!=") | L("~=") | L(">") | L("<")
 116 )
 117
 118 MARKER_OP = VERSION_CMP | L("not in") | L("in")
 119 MARKER_OP.setParseAction(lambda s, l, t: Op(t[0]))
 120
 121 MARKER_VALUE = QuotedString("'") | QuotedString('"')
 122 MARKER_VALUE.setParseAction(lambda s, l, t: Value(t[0]))
 123
 124 BOOLOP = L("and") | L("or")
 125
 126 MARKER_VAR = VARIABLE | MARKER_VALUE
 127
 128 MARKER_ITEM = Group(MARKER_VAR + MARKER_OP + MARKER_VAR)
 129 MARKER_ITEM.setParseAction(lambda s, l, t: tuple(t[0]))
 130
 131 LPAREN = L("(").suppress()
 132 RPAREN = L(")").suppress()
 133
 134 MARKER_EXPR = Forward()
 135 MARKER_ATOM = MARKER_ITEM | Group(LPAREN + MARKER_EXPR + RPAREN)
 136 MARKER_EXPR << MARKER_ATOM + ZeroOrMore(BOOLOP + MARKER_EXPR)
 137
 138 MARKER = stringStart + MARKER_EXPR + stringEnd
 139
 140
 141 def _coerce_parse_result(results: Union[ParseResults, List[Any]]) -> List[Any]:
 142     if isinstance(results, ParseResults):
 143         return [_coerce_parse_result(i) for i in results]
 144     else:
 145         return results
 146
 147
 148 def _format_marker(
 149     marker: Union[List[str], Tuple[Node, ...], str], first: Optional[bool] = True
 150 ) -> str:
 151
 152     assert isinstance(marker, (list, tuple, str))
 153
 154     # Sometimes we have a structure like [[...]] which is a single item list
 155     # where the single item is itself it's own list. In that case we want skip
 156     # the rest of this function so that we don't get extraneous () on the
 157     # outside.
 158     if (
 159         isinstance(marker, list)
 160         and len(marker) == 1
 161         and isinstance(marker[0], (list, tuple))
 162     ):
 163         return _format_marker(marker[0])
 164
 165     if isinstance(marker, list):
 166         inner = (_format_marker(m, first=False) for m in marker)
 167         if first:
 168             return " ".join(inner)
 169         else:
 170             return "(" + " ".join(inner) + ")"
 171     elif isinstance(marker, tuple):
 172         return " ".join([m.serialize() for m in marker])
 173     else:
 174         return marker
 175
 176
 177 _operators: Dict[str, Operator] = {
 178     "in": lambda lhs, rhs: lhs in rhs,
 179     "not in": lambda lhs, rhs: lhs not in rhs,
 180     "<": operator.lt,
 181     "<=": operator.le,
 182     "==": operator.eq,
 183     "!=": operator.ne,
 184     ">=": operator.ge,
 185     ">": operator.gt,
 186 }
 187
 188
 189 def _eval_op(lhs: str, op: Op, rhs: str) -> bool:
 190     try:
 191         spec = Specifier("".join([op.serialize(), rhs]))
 192     except InvalidSpecifier:
 193         pass
 194     else:
 195         return spec.contains(lhs)
 196
 197     oper: Optional[Operator] = _operators.get(op.serialize())
 198     if oper is None:
 199         raise UndefinedComparison(f"Undefined {op!r} on {lhs!r} and {rhs!r}.")
 200
 201     return oper(lhs, rhs)
 202
 203
 204 class Undefined:
 205     pass
 206
 207
 208 _undefined = Undefined()
 209
 210
 211 def _get_env(environment: Dict[str, str], name: str) -> str:
 212     value: Union[str, Undefined] = environment.get(name, _undefined)
 213
 214     if isinstance(value, Undefined):
 215         raise UndefinedEnvironmentName(
 216             f"{name!r} does not exist in evaluation environment."
 217         )
 218
 219     return value
 220
 221
 222 def _evaluate_markers(markers: List[Any], environment: Dict[str, str]) -> bool:
 223     groups: List[List[bool]] = [[]]
 224
 225     for marker in markers:
 226         assert isinstance(marker, (list, tuple, str))
 227
 228         if isinstance(marker, list):
 229             groups[-1].append(_evaluate_markers(marker, environment))
 230         elif isinstance(marker, tuple):
 231             lhs, op, rhs = marker
 232
 233             if isinstance(lhs, Variable):
 234                 lhs_value = _get_env(environment, lhs.value)
 235                 rhs_value = rhs.value
 236             else:
 237                 lhs_value = lhs.value
 238                 rhs_value = _get_env(environment, rhs.value)
 239
 240             groups[-1].append(_eval_op(lhs_value, op, rhs_value))
 241         else:
 242             assert marker in ["and", "or"]
 243             if marker == "or":
 244                 groups.append([])
 245
 246     return any(all(item) for item in groups)
 247
 248
 249 def format_full_version(info: "sys._version_info") -> str:
 250     version = "{0.major}.{0.minor}.{0.micro}".format(info)
 251     kind = info.releaselevel
 252     if kind != "final":
 253         version += kind[0] + str(info.serial)
 254     return version
 255
 256
 257 def default_environment() -> Dict[str, str]:
 258     iver = format_full_version(sys.implementation.version)
 259     implementation_name = sys.implementation.name
 260     return {
 261         "implementation_name": implementation_name,
 262         "implementation_version": iver,
 263         "os_name": os.name,
 264         "platform_machine": platform.machine(),
 265         "platform_release": platform.release(),
 266         "platform_system": platform.system(),
 267         "platform_version": platform.version(),
 268         "python_full_version": platform.python_version(),
 269         "platform_python_implementation": platform.python_implementation(),
 270         "python_version": ".".join(platform.python_version_tuple()[:2]),
 271         "sys_platform": sys.platform,
 272     }
 273
 274
 275 class Marker:
 276     def __init__(self, marker: str) -> None:
 277         try:
 278             self._markers = _coerce_parse_result(MARKER.parseString(marker))
 279         except ParseException as e:
 280             raise InvalidMarker(
 281                 f"Invalid marker: {marker!r}, parse error at "
 282                 f"{marker[e.loc : e.loc + 8]!r}"
 283             )
 284
 285     def __str__(self) -> str:
 286         return _format_marker(self._markers)
 287
 288     def __repr__(self) -> str:
 289         return f"<Marker('{self}')>"
 290
 291     def evaluate(self, environment: Optional[Dict[str, str]] = None) -> bool:
 292         """Evaluate a marker.
 293
 294         Return the boolean from evaluating the given marker against the
 295         environment. environment is an optional argument to override all or
 296         part of the determined environment.
 297
 298         The environment is determined from the current Python process.
 299         """
 300         current_environment = default_environment()
 301         if environment is not None:
 302             current_environment.update(environment)
 303
 304         return _evaluate_markers(self._markers, current_environment)