Source code for pex.translator

# Copyright 2014 Pants project contributors (see
# Licensed under the Apache License, Version 2.0 (see LICENSE).

from __future__ import absolute_import

import os
import traceback
from abc import abstractmethod

from .archiver import Archiver
from .common import chmod_plus_w, safe_copy, safe_mkdtemp, safe_rmtree
from .compatibility import AbstractClass
from .installer import WheelInstaller
from .interpreter import PythonInterpreter
from .package import EggPackage, Package, SourcePackage, WheelPackage
from .pep425tags import get_supported
from .tracer import TRACER
from .util import DistributionHelper

[docs]class TranslatorBase(AbstractClass): """ Translate a link into a distribution. """ @abstractmethod def translate(self, link, into=None): pass
[docs]class ChainedTranslator(TranslatorBase): """ Glue a sequence of Translators together in priority order. The first Translator to resolve a requirement wins. """ def __init__(self, *translators): self._translators = list(filter(None, translators)) for tx in self._translators: if not isinstance(tx, TranslatorBase): raise ValueError('Expected a sequence of translators, got %s instead.' % type(tx)) def translate(self, package, into=None): for tx in self._translators: dist = tx.translate(package, into=into) if dist: return dist def __str__(self): return 'ChainedTranslator(%s)' % ( ', '.join((tx.__class__.__name__ for tx in self._translators)))
class SourceTranslator(TranslatorBase): @classmethod def run_2to3(cls, path): from lib2to3.refactor import get_fixers_from_package, RefactoringTool rt = RefactoringTool(get_fixers_from_package('lib2to3.fixes')) with TRACER.timed('Translating %s' % path): for root, dirs, files in os.walk(path): for fn in files: full_fn = os.path.join(root, fn) if full_fn.endswith('.py'): with TRACER.timed('%s' % fn, V=3): try: chmod_plus_w(full_fn) rt.refactor_file(full_fn, write=True) except IOError: TRACER.log('Failed to translate %s' % fn) TRACER.log(traceback.format_exc()) def __init__(self, interpreter=PythonInterpreter.get(), supported_tags=None, use_2to3=False, installer_impl=WheelInstaller): self._supported_tags = supported_tags or get_supported() self._interpreter = interpreter self._installer_impl = installer_impl self._use_2to3 = use_2to3 def translate(self, package, into=None): """From a SourcePackage, translate to a binary distribution.""" if not isinstance(package, SourcePackage): return None if not package.local: raise ValueError('SourceTranslator cannot translate remote packages.') installer = None version = self._interpreter.version unpack_path = Archiver.unpack(package.local_path) into = into or safe_mkdtemp() try: if self._use_2to3 and version >= (3,): with TRACER.timed('Translating 2->3 %s' % self.run_2to3(unpack_path) installer = self._installer_impl( unpack_path, interpreter=self._interpreter, strict=( not in ('distribute', 'setuptools'))) with TRACER.timed('Packaging %s' % try: dist_path = installer.bdist() except self._installer_impl.InstallFailure as e: TRACER.log('Failed to install package at %s: %s' % (unpack_path, e)) return None target_path = os.path.join(into, os.path.basename(dist_path)) safe_copy(dist_path, target_path) target_package = Package.from_href(target_path) if not target_package: TRACER.log('Target path %s does not look like a Package.' % target_path) return None if not target_package.compatible(self._supported_tags): TRACER.log('Target package %s is not compatible with %s' % ( target_package, self._supported_tags)) return None return DistributionHelper.distribution_from_path(target_path) except Exception as e: TRACER.log('Failed to translate %s' % package) TRACER.log(traceback.format_exc()) finally: if installer: installer.cleanup() if unpack_path: safe_rmtree(unpack_path) class BinaryTranslator(TranslatorBase): def __init__(self, package_type, supported_tags=None): self._package_type = package_type self._supported_tags = supported_tags or get_supported() def translate(self, package, into=None): """From a binary package, translate to a local binary distribution.""" if not package.local: raise ValueError('BinaryTranslator cannot translate remote packages.') if not isinstance(package, self._package_type): return None if not package.compatible(self._supported_tags): TRACER.log('Target package %s is not compatible with %s' % ( package, self._supported_tags)) return None into = into or safe_mkdtemp() target_path = os.path.join(into, package.filename) safe_copy(package.local_path, target_path) return DistributionHelper.distribution_from_path(target_path) class EggTranslator(BinaryTranslator): def __init__(self, **kw): super(EggTranslator, self).__init__(EggPackage, **kw) class WheelTranslator(BinaryTranslator): def __init__(self, **kw): super(WheelTranslator, self).__init__(WheelPackage, **kw) class Translator(object): @staticmethod def default(interpreter=None, supported_tags=None): interpreter = interpreter or PythonInterpreter.get() supported_tags = supported_tags or get_supported() whl_translator = WheelTranslator(supported_tags=supported_tags) egg_translator = EggTranslator(supported_tags=supported_tags) source_translator = SourceTranslator(interpreter=interpreter) return ChainedTranslator(whl_translator, egg_translator, source_translator)