python-update-packages.py

import json
import sys
from datetime import datetime
from functools import lru_cache

import humanize
import requests
from typing import NamedTuple, Dict

import semver
from tabulate import tabulate
from tqdm import tqdm

CONSTRAINT_MASTER_URL = "https://raw.githubusercontent.com/apache/airflow/constraints-master/constraints-3.8.txt"
CONSTRAINT_1_10_URL = "https://raw.githubusercontent.com/apache/airflow/constraints-1.10.14/constraints-3.8.txt"


class PackageInfo(NamedTuple):
    package_name: str

    ver_master: str
    release_data_master: datetime

    ver_1_10: str
    release_data_1_10: datetime

    ver_latest: str
    release_data_latest: datetime

    @property
    @lru_cache(maxsize=None)
    def diff_part(self):
        try:
            current_sem = semver.VersionInfo.parse(self.ver_master)
            latest_sem = semver.VersionInfo.parse(self.ver_latest)
        except ValueError:
            if self.ver_master == self.ver_latest:
                return None
            return "0-unknown"
        if current_sem.major != latest_sem.major:
            return '1-major'
        if current_sem.minor != latest_sem.minor:
            return '2-minor'
        if current_sem.patch != latest_sem.patch:
            return '3-patch'
        return None

    @property
    @lru_cache(maxsize=None)
    def to_update(self):
        try:
            sem_master = semver.VersionInfo.parse(self.ver_master)
            sem_latest = semver.VersionInfo.parse(self.ver_latest)
        except ValueError:
            if self.ver_master == self.ver_latest:
                return "👎🏻"
            return '❓'

        if sem_master.major == sem_latest.major:
            return "👎🏻 No (small difference to latest)"

        if not self.ver_1_10:
            return '👍🏻 Yes (Missing in Airflow v1.10)'

        sem_1_10 = semver.VersionInfo.parse(self.ver_1_10)
        if sem_latest.major != sem_1_10.major:
            return f'👎🏻 No (Conflict - {self.ver_latest} !~= {self.ver_1_10})'
        if sem_master.major != sem_1_10.major:
            return f'❌ Broken'
        return "👍🏻 No"

    @property
    @lru_cache(maxsize=None)
    def age(self):
        if not self.release_data_master or not self.release_data_latest:
            return None
        return self.release_data_master - self.release_data_latest

    def to_dict(self):
        age_latest = humanize.naturaldelta(datetime.now() - self.release_data_latest)
        age_master = humanize.naturaldelta(datetime.now() - self.release_data_master)
        age_1_10 = humanize.naturaldelta(datetime.now() - self.release_data_1_10) if self.release_data_1_10 else None

        return {
            "package_name": self.package_name,
            "master_version": f"{self.ver_master} ({age_master} ago)",
            "1_10_version": f"{self.ver_1_10} ({age_1_10} ago)"  if self.release_data_1_10 else None,
            "latest_version": f"{self.ver_latest} ({age_latest} ago)",
            # "diff_part": self.diff_part,
            "diff between releases": humanize.naturaldelta(self.age),
            "to updaate?": self.to_update
        }


def get_all_packages(constraint_url: str) -> Dict[str, str]:
    request = requests.get(constraint_url)
    request.raise_for_status()
    # Skip empty lines and comments
    all_packages_line_list = (
        l
        for l in request.text.split("\n")
        if l and not l.startswith("#")
    )
    all_packages = {
        p.split("=")[0]: p.split("=")[2]
        for p in all_packages_line_list
    }
    return all_packages


def get_package_metadata(package_name):
    request = requests.get(f"https://pypi.org/pypi/{package_name}/json")
    request.raise_for_status()
    try:
        json_response = request.json()
        return json_response
    except json.decoder.JSONDecodeError:
        print(f"Error fetching packege metadata: {package_name}", file=sys.stderr)
        print(f"Response: {request.text}", file=sys.stderr)

        raise


def parse_pip_date(date_string):
    return datetime.strptime(date_string, "%Y-%m-%dT%H:%M:%S")


def fetch_packages_info(packages_master, packages_1_10):
    results = []
    for package_name, ver_master in tqdm(list(packages_master.items())):
        package_metadata = get_package_metadata(package_name)
        ver_latest = package_metadata['info']['version']
        ver_1_10 = packages_1_10.get(package_name)

        releases = package_metadata["releases"]
        release_data_latest = parse_pip_date(
            releases[ver_latest][0]["upload_time"]
        ) if ver_latest in releases else None
        release_data_1_10 = parse_pip_date(
            releases[ver_1_10][0]["upload_time"]
        ) if ver_1_10 in releases else None
        release_data_master = parse_pip_date(
            releases[ver_master][0]["upload_time"]
        ) if ver_master in releases else None

        results.append(
            PackageInfo(
                package_name=package_name,
                ver_1_10=ver_1_10,
                release_data_1_10=release_data_1_10,
                ver_latest=ver_latest,
                release_data_latest=release_data_latest,
                ver_master=ver_master,
                release_data_master=release_data_master,
            )
        )
    return results


packages_master = get_all_packages(CONSTRAINT_MASTER_URL)
packages_1_10 = get_all_packages(CONSTRAINT_1_10_URL)

filter_key = 'google-cloud'

if filter_key:
    packages_master = {
        package_name: current_version
        for package_name, current_version in packages_master.items()
        if filter_key in package_name
    }

package_infos = fetch_packages_info(packages_master, packages_1_10)

package_infos = [
    p
    for p in package_infos
    if p.diff_part != None
]

sort_by_diff_part = lambda p: (p.diff_part, p.package_name)
sort_by_diff = lambda p: (p.age, p.package_name)
sort_by_age = lambda p: (p.release_data_master, p.package_name)
sort_by_update_to = lambda p: (p.to_update, p.release_data_master, p.package_name)
sort_by = sort_by_update_to

markdown = tabulate(
    tabular_data=[
        d.to_dict()
        for d in sorted(package_infos, key=sort_by)
    ],
    headers="keys", tablefmt="github"
)
print(markdown)
# from pprint import pprint
# pprint(google_cloud_packages)