Added a download_package() method

Author: jwodder

CHANGELOG.md

@@ -18,6 +18,7 @@ v0.10.0 (in development)
       parsing HTML responses
 - Warn on encountering a repository version with a greater minor version than
   expected
+- Gave `PyPISimple` a `download_package()` method
 
 v0.9.0 (2021-08-26)
 -------------------

README.rst

@@ -28,9 +28,9 @@ specified in :pep:`503` and updated by :pep:`592`, :pep:`629`, :pep:`658`, and
 :pep:`691`.  With it, you can query `the Python Package Index (PyPI)
 <https://pypi.org>`_ and other `pip <https://pip.pypa.io>`_-compatible
 repositories for a list of their available projects and lists of each project's
-available package files.  The library also allows you to query package files
-for their project version, package type, file digests, ``requires_python``
-string, PGP signature URL, and metadata URL.
+available package files.  The library also allows you to download package files
+and query them for their project version, package type, file digests,
+``requires_python`` string, PGP signature URL, and metadata URL.
 
 See `the documentation <https://pypi-simple.readthedocs.io>`_ for more
 information.

docs/changelog.rst

@@ -29,6 +29,7 @@ v0.10.0 (in development)
 
 - Warn on encountering a repository version with a greater minor version than
   expected
+- Gave `PyPISimple` a `~PyPISimple.download_package()` method
 
 
 v0.9.0 (2021-08-26)

docs/high-level-api.rst

@@ -6,10 +6,20 @@ High-Level API
 .. autoclass:: IndexPage()
 .. autoclass:: ProjectPage()
 .. autoclass:: DistributionPackage()
+
+Constants
+---------
 .. autodata:: PYPI_SIMPLE_ENDPOINT
 .. autodata:: SUPPORTED_REPOSITORY_VERSION
-.. autoexception:: UnsupportedContentTypeError
+
+Exceptions
+----------
+.. autoexception:: DigestMismatchError()
+    :show-inheritance:
+.. autoexception:: NoDigestsError()
+    :show-inheritance:
+.. autoexception:: UnsupportedContentTypeError()
     :show-inheritance:
-.. autoexception:: UnsupportedRepoVersionError
-.. autoexception:: UnexpectedRepoVersionWarning
+.. autoexception:: UnsupportedRepoVersionError()
+.. autoexception:: UnexpectedRepoVersionWarning()
     :show-inheritance:

docs/index.rst

@@ -22,9 +22,9 @@ specified in :pep:`503` and updated by :pep:`592`, :pep:`629`, :pep:`658`, and
 :pep:`691`.  With it, you can query `the Python Package Index (PyPI)
 <https://pypi.org>`_ and other `pip <https://pip.pypa.io>`_-compatible
 repositories for a list of their available projects and lists of each project's
-available package files.  The library also allows you to query package files
-for their project version, package type, file digests, ``requires_python``
-string, PGP signature URL, and metadata URL.
+available package files.  The library also allows you to download package files
+and query them for their project version, package type, file digests,
+``requires_python`` string, PGP signature URL, and metadata URL.
 
 Installation
 ============

src/pypi_simple/__init__.py

@@ -41,15 +41,19 @@
 )
 from .parse_stream import parse_links_stream, parse_links_stream_response
 from .util import (
+    DigestMismatchError,
+    NoDigestsError,
     UnexpectedRepoVersionWarning,
     UnsupportedContentTypeError,
     UnsupportedRepoVersionError,
 )
 
 __all__ = [
+    "DigestMismatchError",
     "DistributionPackage",
     "IndexPage",
     "Link",
+    "NoDigestsError",
     "PYPI_SIMPLE_ENDPOINT",
     "ProjectPage",
     "PyPISimple",

src/pypi_simple/client.py

@@ -1,12 +1,15 @@
+import os
+from pathlib import Path
 import platform
-from typing import Any, Iterator, List, Optional, Tuple, Union
+from typing import Any, AnyStr, Iterator, List, Optional, Tuple, Union
 from warnings import warn
 from packaging.utils import canonicalize_name as normalize
 import requests
 from . import PYPI_SIMPLE_ENDPOINT, __url__, __version__
 from .classes import DistributionPackage, IndexPage, ProjectPage
 from .parse_repo import parse_repo_index_response, parse_repo_project_response
 from .parse_stream import parse_links_stream_response
+from .util import AbstractDigestChecker, DigestChecker, NullDigestChecker
 
 #: The User-Agent header used for requests; not used when the user provides eir
 #: own session object
@@ -206,6 +209,65 @@ def get_project_url(self, project: str) -> str:
         """
         return self.endpoint + normalize(project) + "/"
 
+    def download_package(
+        self,
+        pkg: DistributionPackage,
+        path: Union[AnyStr, "os.PathLike[AnyStr]"],
+        verify: bool = True,
+        keep_on_error: bool = False,
+        timeout: Union[float, Tuple[float, float], None] = None,
+    ) -> None:
+        """
+        .. versionadded:: 0.10.0
+
+        Download the given `DistributionPackage` to the given path.
+
+        If an error occurs while downloading or verifying digests, and
+        ``keep_on_error`` is not true, the downloaded file is not saved.
+
+        :param DistributionPackage pkg: the distribution package to download
+        :param path:
+            the path at which to save the downloaded file; any parent
+            directories of this path will be created as needed
+        :param bool verify:
+            whether to verify the package's digests against the downloaded file
+        :param bool keep_on_error:
+            whether to keep (true) or delete (false) the downloaded file if an
+            error occurs
+        :param timeout: optional timeout to pass to the ``requests`` call
+        :type timeout: Union[float, Tuple[float,float], None]
+        :raises requests.HTTPError: if the repository responds with an HTTP
+            error code
+        :raises NoDigestsError:
+            if ``verify`` is true and the given package does not have any
+            digests with known algorithms
+        :raises DigestMismatchError:
+            if ``verify`` is true and the digest of the downloaded file does
+            not match the expected value
+        """
+        target = Path(os.fsdecode(path))
+        target.parent.mkdir(parents=True, exist_ok=True)
+        digester: AbstractDigestChecker
+        if verify:
+            digester = DigestChecker(pkg.digests)
+        else:
+            digester = NullDigestChecker()
+        with self.s.get(pkg.url, stream=True, timeout=timeout) as r:
+            r.raise_for_status()
+            try:
+                with target.open("wb") as fp:
+                    for chunk in r.iter_content(65535):
+                        fp.write(chunk)
+                        digester.update(chunk)
+                digester.finalize()
+            except Exception:
+                if not keep_on_error:
+                    try:
+                        target.unlink()
+                    except FileNotFoundError:
+                        pass
+                raise
+
     def get_projects(self) -> Iterator[str]:
         """
         Returns a generator of names of projects available in the repository.

src/pypi_simple/util.py

@@ -1,4 +1,6 @@
-from typing import Optional
+from abc import ABC, abstractmethod
+import hashlib
+from typing import Any, Dict, Optional
 from urllib.parse import urljoin
 import warnings
 from packaging.version import Version
@@ -86,3 +88,87 @@ def basejoin(base_url: Optional[str], url: str) -> str:
         return url
     else:
         return urljoin(base_url, url)
+
+
+class AbstractDigestChecker(ABC):
+    @abstractmethod
+    def update(self, blob: bytes) -> None:
+        ...
+
+    @abstractmethod
+    def finalize(self) -> None:
+        pass
+
+
+class NullDigestChecker(AbstractDigestChecker):
+    def update(self, blob: bytes) -> None:
+        pass
+
+    def finalize(self) -> None:
+        pass
+
+
+class DigestChecker(AbstractDigestChecker):
+    def __init__(self, digests: Dict[str, str]) -> None:
+        self.digesters: Dict[str, Any] = {}
+        self.expected: Dict[str, str] = {}
+        for alg, value in digests.items():
+            try:
+                d = hashlib.new(alg)
+            except ValueError:
+                pass
+            else:
+                self.digesters[alg] = d
+                self.expected[alg] = value
+        if not self.digesters:
+            raise NoDigestsError("No digests with known algorithms available")
+
+    def update(self, blob: bytes) -> None:
+        for d in self.digesters.values():
+            d.update(blob)
+
+    def finalize(self) -> None:
+        for alg, d in self.digesters.items():
+            actual = d.hexdigest()
+            if actual != self.expected[alg]:
+                raise DigestMismatchError(
+                    algorithm=alg,
+                    expected_digest=self.expected[alg],
+                    actual_digest=actual,
+                )
+
+
+class NoDigestsError(ValueError):
+    """
+    .. versionadded:: 0.10.0
+
+    Raised by `PyPISimple.download_package()` with ``verify=True`` when the
+    given package does not have any digests with known algorithms
+    """
+
+    pass
+
+
+class DigestMismatchError(ValueError):
+    """
+    .. versionadded:: 0.10.0
+
+    Raised by `PyPISimple.download_package()` with ``verify=True`` when the
+    digest of the downloaded file does not match the expected value
+    """
+
+    def __init__(
+        self, algorithm: str, expected_digest: str, actual_digest: str
+    ) -> None:
+        #: The name of the digest algorithm used
+        self.algorithm = algorithm
+        #: The expected digest
+        self.expected_digest = expected_digest
+        #: The digest of the file that was actually received
+        self.actual_digest = actual_digest
+
+    def __str__(self) -> str:
+        return (
+            f"{self.algorithm} digest of downloaded file is"
+            f" {self.actual_digest!r} instead of expected {self.expected_digest!r}"
+        )