Pants task for determining pypi package staleness

howstale.py

from __future__ import (nested_scopes, generators, division, absolute_import, with_statement,
                        print_function, unicode_literals)

import datetime
from distutils import version as disutils_version  # pylint: disable=no-name-in-module

import requests
from pants.backend.python.targets import python_requirement_library  # pylint: disable=no-name-in-module
from pants.task import console_task
from pants.util import memo

_ISO_FORMAT_STR = '%Y-%m-%dT%H:%M:%S'


class HowStale(console_task.ConsoleTask):
    """[Oscar] Looks for python requirements in the given context and outputs how stale they are versus PyPI"""

    @classmethod
    def register_options(cls, register):
        super(HowStale, cls).register_options(register)
        register('--ignore-prerelease', type=bool, help='If true, ignores pre-release packages in PyPI', default=True)
        register('--pypi-url-pattern', type=str, help='Url pattern to use for fetching python package info',
                 default='https://pypi.python.org/pypi/{package}/json')

    def console_output(self, _):
        for target in self.context.targets(
                lambda t: isinstance(t, python_requirement_library.PythonRequirementLibrary)
        ):
            for python_req in target.requirements:
                requirement = python_req.requirement
                if 'oscar' in requirement.name.lower():
                    self.context.log.warn(
                        'Skipping oscar specific package {}'.format(requirement.name)
                    )
                    continue
                if len(requirement.specs) != 1:
                    self.context.log.warn(
                        'Skipping {}, does not specify a single requirement spec'.format(requirement.name)
                    )
                specifier, version = requirement.specs[0]
                if not specifier == '==':
                    self.context.log.warn(
                        'Skipping {}, does not pin the package version'.format(requirement.name)
                    )
                    continue
                try:
                    package_diff = self._compare_package_with_latest(requirement.name, version)
                    yield str(package_diff)
                except Exception as exc:  # pylint: disable=broad-except
                    self.context.log.error(
                        'Error fetching package info for {}: exc: {}'.format(requirement.name, exc.message)
                    )

    @memo.memoized_method
    def _get_package_info(self, package):
        """Fetches a package's info json from pypi and returns the parsed json
        :param str package: package name
        :rtype: dict
        """
        resp = requests.get(self.get_options().pypi_url_pattern.format(package=package))
        if resp.ok:
            return resp.json()

    def _compare_package_with_latest(self, package, version):
        """Returns a package_difference object representing the difference between the supplied
         package and version and the package's latest version.

         Only supports versions using semantic versioning
        :param str package:
        :param str version:
        :rtype package_difference instance
        """
        package_diff = _PackageDifference(package, version)

        package_info = self._get_package_info(package)
        latest_release_version = package_info['info']['version']
        package_diff.latest_package_version = latest_release_version

        latest_release_info = package_info['releases'][latest_release_version][0]
        current_release_info = package_info['releases'][version][0]

        # compute time diff
        latest_release_date = datetime.datetime.strptime(
            latest_release_info['upload_time'], _ISO_FORMAT_STR)
        current_release_date = datetime.datetime.strptime(
            current_release_info['upload_time'], _ISO_FORMAT_STR)
        package_diff.time_behind = latest_release_date - current_release_date

        # compute version diff
        try:
            latest_major, latest_minor, latest_patch = disutils_version.StrictVersion(latest_release_version).version
            current_major, current_minor, current_patch = disutils_version.StrictVersion(version).version
            if latest_major > current_major:
                package_diff.major_version_behind = latest_major - current_major
            elif latest_minor > current_minor:
                package_diff.minor_version_behind = latest_minor - current_minor
            elif latest_patch > current_patch:
                package_diff.patch_version_behind = latest_patch - current_patch
        except ValueError:
            pass

        return package_diff


class _PackageDifference(object):
    def __init__(
            self, package, version, time_behind=None, latest_package_version=None, major_version_behind=None,
            minor_version_behind=None,
            patch_version_behind=None
    ):
        self.package = package
        self.version = version
        self.latest_package_version = latest_package_version
        self.time_behind = time_behind
        self.major_version_behind = major_version_behind
        self.minor_version_behind = minor_version_behind
        self.patch_version_behind = patch_version_behind

    @property
    def up_to_date(self):
        return self.version == self.latest_package_version

    def __str__(self):
        header = 'Package {} version {}:\n'.format(self.package, self.version)
        lines = []
        if self.up_to_date:
            lines.append('Up to date')
        else:
            lines.append('Latest version: {}'.format(self.latest_package_version))
            if self.time_behind:
                lines.append('{} days older than latest version'.format(self.time_behind.days))
            if self.major_version_behind:
                lines.append('{} major versions behind latest'.format(self.major_version_behind))
            elif self.minor_version_behind:
                lines.append('{} minor versions behind latest'.format(self.minor_version_behind))
            elif self.patch_version_behind:
                lines.append('{} patch versions behind latest'.format(self.patch_version_behind))
        return header + '\n'.join(['\t{}'.format(line) for line in lines])