Source code for pip_rating.sources.pypi

import datetime
from functools import cached_property
from itertools import chain
from typing import TypedDict, Optional, List, Dict

import requests

from pip_rating.sources.base import SourceBase

URL = "https://pypi.org/pypi/{package_name}/json"


class PyPackageInfo(TypedDict):
    author: str
    author_email: str
    bugtrack_url: str
    classifiers: List[str]
    description: str
    description_content_type: str
    docs_url: str
    download_url: str
    downloads: Dict[str, int]
    home_page: str
    keywords: str
    license: str
    maintainer: str
    maintainer_email: str
    name: str
    package_url: str
    platform: Optional[str]
    project_url: str
    project_urls: Dict[str, str]
    release_url: str
    requires_dist: List[str]
    requires_python: Optional[str]
    summary: str
    version: str
    yanked: bool
    yanked_reason: Optional[str]


class PypiPackageReleaseUpload(TypedDict):
    comment_text: str
    digests: Dict[str, str]
    downloads: int
    filename: str
    has_sig: bool
    md5_digest: str
    packagetype: str
    python_version: str
    requires_python: Optional[str]
    size: int
    upload_time: str
    upload_time_iso_8601: str
    url: str
    yanked: bool
    yanked_reason: Optional[str]


class PypiPackage(TypedDict):
    info: PyPackageInfo
    last_serial: int
    releases: Dict[str, List[PypiPackageReleaseUpload]]


class PypiCacheDict(TypedDict):
    package_name: str
    updated_at: str
    package: PypiPackage


class Pypi(SourceBase):
    source_name = "pypi"

    def get_cache_data(self) -> dict:
        return {
            "package_name": self.package_name,
            "updated_at": datetime.datetime.now().isoformat(),
            "package": dict(self.get_package()),
        }

    @cached_property
    def package(self) -> PypiPackage:
        if not self.is_cache_expired:
            cache = self.get_from_cache()
        else:
            cache = self.save_to_cache()
        return cache["package"]

    @cached_property
    def uploads(self) -> List[PypiPackageReleaseUpload]:
        uploads = chain(*list(self.package["releases"].values()))
        return sorted(uploads, key=lambda upload: upload.get("upload_time_iso_8601"))

    @property
    def latest_upload(self) -> Optional[PypiPackageReleaseUpload]:
        return self.uploads[-1] if self.uploads else None

    @property
    def first_upload(self) -> Optional[PypiPackageReleaseUpload]:
        return self.uploads[0] if self.uploads else None

    @property
    def latest_upload_iso_dt(self) -> Optional[str]:
        if self.latest_upload:
            return self.latest_upload["upload_time_iso_8601"]
        return None

    @property
    def first_upload_iso_dt(self) -> Optional[str]:
        if self.first_upload:
            return self.first_upload["upload_time_iso_8601"]
        return None

    def get_package(self) -> PypiPackage:
        with requests.get(URL.format(package_name=self.package_name)) as response:
            response.raise_for_status()
            return response.json()