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_URL = "https://raw.githubusercontent.com/apache/airflow/constraints-master/constraints-3.8.txt"


class PackageInfo(NamedTuple):
    package_name: str
    current_version: str
    latest_version: str
    current_version_release_data: datetime
    latest_version_release_data: datetime

    @property
    @lru_cache(maxsize=None)
    def diff_part(self):
        try:
            current_sem = semver.VersionInfo.parse(self.current_version)
            latest_sem = semver.VersionInfo.parse(self.latest_version)
        except ValueError:
            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 age(self):
        if not self.current_version_release_data or not self.latest_version_release_data:
            return None
        return self.current_version_release_data-self.latest_version_release_data

    def to_dict(self):
        age_latest = humanize.naturaldelta(datetime.now() - self.latest_version_release_data)
        age_current = humanize.naturaldelta(datetime.now() - self.current_version_release_data)

        return {
            "package_name": self.package_name,
            "current_version": f"{self.current_version} ({age_current} ago)",
            "latest_version": f"{self.latest_version} ({age_latest} ago)",
            "diff_part": self.diff_part,
            "diff between releases": humanize.naturaldelta(self.age)
        }


def get_all_packages() -> 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 metadatan: {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):
    results = []
    for package_name, current_version in tqdm(list(packages.items())):
        package_metadata = get_package_metadata(package_name)
        latest_version = package_metadata['info']['version']

        releases = package_metadata["releases"]
        current_version_release_data = parse_pip_date(
            releases[current_version][0]["upload_time"]
        ) if current_version in releases else None
        latest_version_release_data = parse_pip_date(
            releases[latest_version][0]["upload_time"]
        ) if latest_version in releases else None

        results.append(
            PackageInfo(
                package_name=package_name,
                current_version=current_version,
                latest_version=latest_version,
                current_version_release_data=current_version_release_data,
                latest_version_release_data=latest_version_release_data,
            )
        )
    return results


all_packages = get_all_packages()
# filter_key = 'google-cloud'
#
# if filter_key:
#     all_packages = {
#         package_name: current_version
#         for package_name, current_version in all_packages.items()
#         if filter_key in package_name
#     }

package_infos = fetch_packages_info(all_packages)

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

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.current_version_release_data, p.package_name)
sort_by = sort_by_age

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