Source code for pex.package

# Copyright 2014 Pants project contributors (see
# 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).""" class Error(Exception): pass 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:, 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( 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 link 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 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. def compatible(self, supported_tags): return True
[docs]class EggPackage(Package): """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 = '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) abi_tag = 'none' tag_platform = self._platform or 'any' self._supported_tags = set() self._supported_tags.add(('py' + self._py_version.replace('.', ''), abi_tag, tag_platform)) # Work around PyPy versions being weird in pep425tags.get_supported if self.py_version == '2.7': self._supported_tags.add(('pp2', abi_tag, tag_platform)) elif self.py_version == '3.2': self._supported_tags.add(('pp321', abi_tag, tag_platform)) elif self.py_version == '3.3': self._supported_tags.add(('pp352', abi_tag, tag_platform)) elif self.py_version == '3.5': self._supported_tags.add(('pp357', abi_tag, tag_platform)) def __hash__(self): return hash((, 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
[docs]class WheelPackage(Package): """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 for why this is unavoidable. self._name = self._name.replace('_', '-') self._raw_version = self._raw_version.replace('_', '-') self._supported_tags = frozenset(self._iter_tags()) @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)