markers.py 8.3 KB

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