Source code for pex.package

# 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)