Skip to content

Commit

Permalink
PEXEnvironment for a DistributionTarget. (#1178)
Browse files Browse the repository at this point in the history
Instead of requiring a PythonIntepreter to resolve distributions in a
PEXEnvironment, a DistributionTarget is enough. This allows for
resolving distributions from a PEXEnvironment given only a Platform
which we need to support resolving from a `--pex-repository` in #1108.
  • Loading branch information
jsirois authored Jan 20, 2021
1 parent ab20677 commit 451eced
Show file tree
Hide file tree
Showing 8 changed files with 229 additions and 146 deletions.
106 changes: 86 additions & 20 deletions pex/distribution_target.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,79 +6,145 @@
import os

from pex.interpreter import PythonInterpreter
from pex.platforms import Platform
from pex.third_party.packaging import tags
from pex.third_party.pkg_resources import Requirement
from pex.typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
from typing import Any, Optional, Tuple


class DistributionTarget(object):
"""Represents the target of a python distribution."""

@classmethod
def current(cls):
return cls(interpreter=None, platform=None)
# type: () -> DistributionTarget
return cls()

@classmethod
def for_interpreter(cls, interpreter):
return cls(interpreter=interpreter, platform=None)
# type: (PythonInterpreter) -> DistributionTarget
return cls(interpreter=interpreter)

@classmethod
def for_platform(cls, platform):
return cls(interpreter=None, platform=platform)

def __init__(self, interpreter=None, platform=None):
def for_platform(
cls,
platform, # type: Platform
manylinux=None, # type: Optional[str]
):
# type: (...) -> DistributionTarget
return cls(platform=platform, manylinux=manylinux)

def __init__(
self,
interpreter=None, # type: Optional[PythonInterpreter]
platform=None, # type:Optional[Platform]
manylinux=None, # type: Optional[str]
):
# type: (...) -> None
if interpreter and platform:
raise ValueError(
"A {class_name} can represent an interpreter or a platform but not both at the "
"same time. Given interpreter {interpreter} and platform {platform}.".format(
class_name=self.__class__.__name__, interpreter=interpreter, platform=platform
)
)
if manylinux and not platform:
raise ValueError(
"A value for manylinux only makes sense for platform distribution targets. Given "
"manylinux={!r} but no platform.".format(manylinux)
)
self._interpreter = interpreter
self._platform = platform
self._manylinux = manylinux

@property
def is_foreign(self):
# type: () -> bool
if self._platform is None:
return False
return self._platform not in self.get_interpreter().supported_platforms

def get_interpreter(self):
# type: () -> PythonInterpreter
return self._interpreter or PythonInterpreter.get()

def get_platform(self):
return self._platform or self.get_interpreter().platform
def get_python_version_str(self):
# type: () -> Optional[str]
if self._platform is not None:
return None
return self.get_interpreter().identity.version_str

def requirement_applies(self, requirement):
def get_platform(self):
# type: () -> Tuple[Platform, Optional[str]]
if self._platform is not None:
return self._platform, self._manylinux
return self.get_interpreter().platform, None

def get_supported_tags(self):
# type: () -> Tuple[tags.Tag, ...]
if self._platform is not None:
return self._platform.supported_tags(manylinux=self._manylinux)
return self.get_interpreter().identity.supported_tags

def requirement_applies(
self,
requirement, # type: Requirement
extras=None, # type: Optional[Tuple[str, ...]]
):
# type: (...) -> bool
"""Determines if the given requirement applies to this distribution target.
:param requirement: The requirement to evaluate.
:type requirement: :class:`pex.third_party.pkg_resources.Requirement`
:rtype: bool
:param extras: Optional active extras.
"""
if not requirement.marker:
if requirement.marker is None:
return True

if self._interpreter is None:
if self._platform is not None:
return True

return requirement.marker.evaluate(self._interpreter.identity.env_markers)
if not extras:
# Provide an empty extra to safely evaluate the markers without matching any extra.
extras = ("",)
for extra in extras:
# N.B.: This nets us a copy of the markers so we're free to mutate.
environment = self.get_interpreter().identity.env_markers
environment["extra"] = extra
if requirement.marker.evaluate(environment=environment):
return True

return False

@property
def id(self):
"""A unique id for this distribution target suitable as a path name component.
:rtype: str
"""
# type: () -> str
"""A unique id for this distribution target suitable as a path name component."""
if self._platform is None:
interpreter = self.get_interpreter()
return interpreter.binary.replace(os.sep, ".").lstrip(".")
else:
return str(self._platform)

def __repr__(self):
# type: () -> str
if self._platform is None:
return "{}(interpreter={!r})".format(self.__class__.__name__, self.get_interpreter())
else:
return "{}(platform={!r})".format(self.__class__.__name__, self._platform)

def _tup(self):
# type: () -> Tuple[Any, ...]
return self._interpreter, self._platform

def __eq__(self, other):
if type(other) is not type(self):
# type: (Any) -> bool
if type(other) is not DistributionTarget:
return NotImplemented
return self._tup() == other._tup()
return self._tup() == cast(DistributionTarget, other)._tup()

def __hash__(self):
# type: () -> int
return hash(self._tup())
Loading

0 comments on commit 451eced

Please sign in to comment.