# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# Licensed under the Apache License, Version 2.0 (see LICENSE).
import os
from pkg_resources import EGG_NAME, parse_version, safe_name, safe_version
from .archiver import Archiver
from .base import maybe_requirement
from .link import Link
from .pep425tags import get_supported
from .util import Memoizer
[docs]class Package(Link):
"""Base class for named Python binary packages (e.g. source, egg, wheel)."""
[docs] class Error(Exception): pass
[docs] class InvalidPackage(Error): pass
# The registry of concrete implementations
_REGISTRY = set()
# The cache of packages that we have already constructed.
_HREF_TO_PACKAGE_CACHE = Memoizer()
[docs] @classmethod
def register(cls, package_type):
"""Register a concrete implementation of a Package to be recognized by pex."""
if not issubclass(package_type, cls):
raise TypeError('package_type must be a subclass of Package.')
cls._REGISTRY.add(package_type)
[docs] @classmethod
def from_href(cls, href, **kw):
"""Convert from a url to Package.
:param href: The url to parse
:type href: string
:returns: A Package object if a valid concrete implementation exists, otherwise None.
"""
package = cls._HREF_TO_PACKAGE_CACHE.get(href)
if package is not None:
return package
link_href = Link.wrap(href)
for package_type in cls._REGISTRY:
try:
package = package_type(link_href.url, **kw)
break
except package_type.InvalidPackage:
continue
if package is not None:
cls._HREF_TO_PACKAGE_CACHE.store(href, package)
return package
@property
def name(self):
return NotImplementedError
@property
def raw_version(self):
return NotImplementedError
@property
def version(self):
return parse_version(self.raw_version)
[docs] def satisfies(self, requirement, allow_prereleases=None):
"""Determine whether this package matches the requirement.
:param requirement: The requirement to compare this Package against
:type requirement: string or :class:`pkg_resources.Requirement`
:param Optional[bool] allow_prereleases: Whether to allow prereleases to satisfy
the `requirement`.
:returns: True if the package matches the requirement, otherwise False
"""
requirement = maybe_requirement(requirement)
link_name = safe_name(self.name).lower()
if link_name != requirement.key:
return False
# NB: If we upgrade to setuptools>=34 the SpecifierSet used here (requirement.specifier) will
# come from a non-vendored `packaging` package and pex's bootstrap code in `PEXBuilder` will
# need an update.
return requirement.specifier.contains(self.raw_version, prereleases=allow_prereleases)
[docs] def compatible(self, supported_tags):
"""Is this package compatible with the given tag set?
:param supported_tags: A list of tags that is supported by the target
interpeter, as generated by
:func:`pex.pep425tags.get_supported`.
:type supported_tags: list of 3-tuples
"""
raise NotImplementedError()
[docs]class SourcePackage(Package):
"""A Package representing an uncompiled/unbuilt source distribution."""
[docs] @classmethod
def split_fragment(cls, fragment):
"""A heuristic used to split a string into version name/fragment:
>>> SourcePackage.split_fragment('pysolr-2.1.0-beta')
('pysolr', '2.1.0-beta')
>>> SourcePackage.split_fragment('cElementTree-1.0.5-20051216')
('cElementTree', '1.0.5-20051216')
>>> SourcePackage.split_fragment('pil-1.1.7b1-20090412')
('pil', '1.1.7b1-20090412')
>>> SourcePackage.split_fragment('django-plugin-2-2.3')
('django-plugin-2', '2.3')
"""
def likely_version_component(enumerated_fragment):
return sum(bool(v and v[0].isdigit()) for v in enumerated_fragment[1].split('.'))
fragments = fragment.split('-')
if len(fragments) == 1:
return fragment, ''
max_index, _ = max(enumerate(fragments), key=likely_version_component)
return '-'.join(fragments[0:max_index]), '-'.join(fragments[max_index:])
def __init__(self, url, **kw):
super(SourcePackage, self).__init__(url, **kw)
ext = Archiver.get_extension(self.filename)
if ext is None:
raise self.InvalidPackage('%s is not a recognized archive format.' % self.filename)
fragment = self.filename[:-len(ext)]
self._name, self._raw_version = self.split_fragment(fragment)
@property
def name(self):
return safe_name(self._name)
@property
def raw_version(self):
return safe_version(self._raw_version)
# SourcePackages are always compatible as they can be translated to a distribution.
[docs] def compatible(self, supported_tags):
return True
[docs]class BinaryPackage(Package):
"""A Package representing a binary distribution which can be e.g. platform specific."""
def __init__(self, *args, **kwargs):
super(BinaryPackage, self).__init__(*args, **kwargs)
self._supported_tags = None
@property
def supported_tags(self):
if not self._supported_tags:
self._supported_tags = frozenset(self._iter_tags())
return self._supported_tags
def _iter_tags(self):
raise NotImplementedError()
[docs] def compatible(self, supported_tags):
"""Is this package compatible with the given tag set?
:param supported_tags: A list of tags that is supported by the target
interpeter, as generated by
:func:`pex.pep425tags.get_supported`.
:type supported_tags: list of 3-tuples
"""
return not set(supported_tags).isdisjoint(self.supported_tags)
[docs]class EggPackage(BinaryPackage):
"""A Package representing a built egg."""
def __init__(self, url, **kw):
super(EggPackage, self).__init__(url, **kw)
filename, ext = os.path.splitext(self.filename)
if ext.lower() != '.egg':
raise self.InvalidPackage('Not an egg: %s' % filename)
matcher = EGG_NAME(filename)
if not matcher:
raise self.InvalidPackage('Could not match egg: %s' % filename)
self._name, self._raw_version, self._py_version, self._platform = matcher.group(
'name', 'ver', 'pyver', 'plat'
)
if self._raw_version is None or self._py_version is None:
raise self.InvalidPackage('url with .egg extension but bad name: %s' % url)
def __hash__(self):
return hash((self.name, self.version, self.py_version, self.platform))
@property
def name(self):
return safe_name(self._name)
@property
def raw_version(self):
return safe_version(self._raw_version)
@property
def py_version(self):
return self._py_version
@property
def platform(self):
return self._platform
def _iter_tags(self):
abi_tag = 'none'
tag_platform = (self._platform or 'any').replace('-', '_').replace('.', '_')
suffix_maj = self._py_version.split('.', 1)[0]
suffix_min = self._py_version.replace('.', '')
for prefix in ('py', 'cp'):
for suffix in (suffix_maj, suffix_min):
impl = ''.join((prefix, suffix))
yield (impl, abi_tag, tag_platform)
# Work around PyPy versions being weird in pep425tags.get_supported
if self.py_version == '2.7':
yield ('pp2', abi_tag, tag_platform)
elif self.py_version == '3.2':
yield ('pp321', abi_tag, tag_platform)
elif self.py_version == '3.3':
# N.B. PyPy 3.3 maps to `pp352` because of PyPy versioning.
# see e.g. http://doc.pypy.org/en/latest/release-pypy3.3-v5.2-alpha1.html
yield ('pp352', abi_tag, tag_platform)
elif self.py_version == '3.5':
yield ('pp357', abi_tag, tag_platform)
[docs]class WheelPackage(BinaryPackage):
"""A Package representing a built wheel."""
def __init__(self, url, **kw):
super(WheelPackage, self).__init__(url, **kw)
filename, ext = os.path.splitext(self.filename)
if ext.lower() != '.whl':
raise self.InvalidPackage('Not a wheel: %s' % filename)
try:
self._name, self._raw_version, self._py_tag, self._abi_tag, self._arch_tag = (
filename.rsplit('-', 4))
except ValueError:
raise self.InvalidPackage('Wheel filename malformed.')
# See https://github.com/pypa/pip/issues/1150 for why this is unavoidable.
self._name = self._name.replace('_', '-')
self._raw_version = self._raw_version.replace('_', '-')
@property
def name(self):
return self._name
@property
def raw_version(self):
return self._raw_version
def _iter_tags(self):
for py in self._py_tag.split('.'):
for abi in self._abi_tag.split('.'):
for arch in self._arch_tag.split('.'):
yield (py, abi, arch)
def __eq__(self, other):
return (self._name == other._name and
self._raw_version == other._raw_version and
self._py_tag == other._py_tag and
self._abi_tag == other._abi_tag and
self._arch_tag == other._arch_tag)
def __hash__(self):
return hash((self._name, self._raw_version, self._py_tag, self._abi_tag, self._arch_tag))
Package.register(SourcePackage)
Package.register(EggPackage)
Package.register(WheelPackage)
[docs]def distribution_compatible(dist, supported_tags=None):
"""Is this distribution compatible with the given interpreter/platform combination?
:param supported_tags: A list of tag tuples specifying which tags are supported
by the platform in question.
:returns: True if the distribution is compatible, False if it is unrecognized or incompatible.
"""
if supported_tags is None:
supported_tags = get_supported()
package = Package.from_href(dist.location)
if not package:
return False
return package.compatible(supported_tags)