Source code for pex.translator

# 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


class TranslatorBase(AbstractClass):
  """
    Translate a link into a distribution.
  """

  @abstractmethod
  def translate(self, link, into=None):
    pass


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)