# Copyright 2014 Pants project contributors (see CONTRIBUTORS.md).
# 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' % package.name):
self.run_2to3(unpack_path)
installer = self._installer_impl(
unpack_path,
interpreter=self._interpreter,
strict=(package.name not in ('distribute', 'setuptools')))
with TRACER.timed('Packaging %s' % package.name):
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)