Add support for PyPI PURLs in purl resolution

Signed-off-by: tdruez <[email protected]>

Author: tdruez

component_catalog/models.py

@@ -54,9 +54,8 @@
 from component_catalog.license_expression_dje import get_license_objects
 from component_catalog.license_expression_dje import parse_expression
 from component_catalog.license_expression_dje import render_expression_as_html
+from dejacode_toolkit import download
 from dejacode_toolkit import spdx
-from dejacode_toolkit.download import DataCollectionException
-from dejacode_toolkit.download import collect_package_data
 from dejacode_toolkit.purldb import PurlDB
 from dejacode_toolkit.purldb import pick_purldb_entry
 from dejacode_toolkit.scancodeio import ScanCodeIO
@@ -2122,8 +2121,8 @@ def collect_data(self, force_update=False, save=True):
             return
 
         try:
-            package_data = collect_package_data(self.download_url)
-        except DataCollectionException as e:
+            package_data = download.collect_package_data(self.download_url)
+        except download.DataCollectionException as e:
             tasks_logger.info(e)
             return
         tasks_logger.info("Package data collected.")
@@ -2476,7 +2475,7 @@ def create_from_url(cls, url, user):
         scoped_packages_qs = cls.objects.scope(user.dataspace)
 
         if is_purl_str(url):
-            download_url = purl2url.get_download_url(url)
+            download_url = download.infer_download_url(url)
             package_url = PackageURL.from_string(url)
             existing_packages = scoped_packages_qs.for_package_url(url, exact_match=True)
         else:
@@ -2504,7 +2503,7 @@ def create_from_url(cls, url, user):
                 package_data.update(purldb_data)
 
         if download_url and not purldb_data:
-            package_data = collect_package_data(download_url)
+            package_data = download.collect_package_data(download_url)
 
         # Check for existing package by hash fields with a single database query
         hash_fields = ["sha512", "sha256", "sha1", "md5"]

component_catalog/tests/test_models.py

@@ -1701,7 +1701,7 @@ def test_package_model_update_from_data(self):
         package.refresh_from_db()
         self.assertEqual("apache-2.0", package.declared_license_expression)
 
-    @mock.patch("component_catalog.models.collect_package_data")
+    @mock.patch("dejacode_toolkit.download.collect_package_data")
     def test_package_model_create_from_url(self, mock_collect):
         self.assertIsNone(Package.create_from_url(url=" ", user=self.user))
 
@@ -1729,6 +1729,15 @@ def test_package_model_create_from_url(self, mock_collect):
         self.assertEqual(purl, package.package_url)
         mock_collect.assert_called_with("https://registry.npmjs.org/is-npm/-/is-npm-1.0.0.tgz")
 
+        purl = "pkg:pypi/[email protected]"
+        download_url = "https://files.pythonhosted.org/packages/Django-5.2.tar.gz"
+        mock_collect.return_value = {}
+        with mock.patch("dejacode_toolkit.download.PyPIFetcher.get_download_url") as mock_pypi_get:
+            mock_pypi_get.return_value = download_url
+            package = Package.create_from_url(url=purl, user=self.user)
+        self.assertEqual(purl, package.package_url)
+        mock_collect.assert_called_with(download_url)
+
     @mock.patch("component_catalog.models.Package.get_purldb_entries")
     @mock.patch("dejacode_toolkit.purldb.PurlDB.is_configured")
     def test_package_model_create_from_url_enable_purldb_access(

component_catalog/tests/test_views.py

@@ -1726,7 +1726,7 @@ def test_package_create_ajax_view(self):
             "sha1": "5ba93c9db0cff93f52b521d7420e43f6eda2784f",
             "md5": "93b885adfe0da089cdf634904fd59f71",
         }
-        with mock.patch("component_catalog.models.collect_package_data") as collect:
+        with mock.patch("dejacode_toolkit.download.collect_package_data") as collect:
             collect.return_value = collected_data
             response = self.client.post(package_add_url, data)
 
@@ -1749,7 +1749,7 @@ def test_package_create_ajax_view(self):
         # Different URL but sha1 match in the db
         data = {"download_urls": "https://url.com/file.ext"}
         collected_data["download_url"] = data["download_urls"]
-        with mock.patch("component_catalog.models.collect_package_data") as collect:
+        with mock.patch("dejacode_toolkit.download.collect_package_data") as collect:
             collect.return_value = collected_data
             response = self.client.post(package_add_url, data)
 

component_catalog/views.py

@@ -47,7 +47,6 @@
 from crispy_forms.utils import render_crispy_form
 from natsort import natsorted
 from packageurl import PackageURL
-from packageurl.contrib import purl2url
 
 from component_catalog.filters import ComponentFilterSet
 from component_catalog.filters import PackageFilterSet
@@ -72,6 +71,7 @@
 from component_catalog.models import PackageAlreadyExistsWarning
 from component_catalog.models import Subcomponent
 from dejacode_toolkit.download import DataCollectionException
+from dejacode_toolkit.download import infer_download_url
 from dejacode_toolkit.purldb import PurlDB
 from dejacode_toolkit.scancodeio import ScanCodeIO
 from dejacode_toolkit.scancodeio import ScanStatus
@@ -1943,7 +1943,7 @@ def get_initial(self):
             purl = PackageURL.from_string(package_url)
             package_url_dict = purl.to_dict(encode=True, empty="")
             initial.update(package_url_dict)
-            if download_url := purl2url.get_download_url(package_url):
+            if download_url := infer_download_url(purl):
                 initial.update({"download_url": download_url})
 
         return initial

dejacode_toolkit/download.py

@@ -6,6 +6,7 @@
 # See https://aboutcode.org for more information about AboutCode FOSS projects.
 #
 
+from contextlib import suppress
 from pathlib import Path
 from urllib.parse import unquote
 from urllib.parse import urlparse
@@ -14,13 +15,16 @@
 from django.utils.http import parse_header_parameters
 
 import requests
+from packageurl import PackageURL
+from packageurl.contrib import purl2url
 
 from dejacode_toolkit.utils import md5
 from dejacode_toolkit.utils import sha1
 from dejacode_toolkit.utils import sha256
 from dejacode_toolkit.utils import sha512
 
 CONTENT_MAX_LENGTH = 536870912  # 512 MB
+DEFAULT_TIMEOUT = 5
 
 
 class DataCollectionException(Exception):
@@ -29,7 +33,7 @@ class DataCollectionException(Exception):
 
 def collect_package_data(url):
     try:
-        response = requests.get(url, timeout=5, stream=True)
+        response = requests.get(url, timeout=DEFAULT_TIMEOUT, stream=True)
     except (TimeoutError, requests.RequestException) as e:
         raise DataCollectionException(e)
 
@@ -73,3 +77,94 @@ def collect_package_data(url):
     }
 
     return package_data
+
+
+class PyPIFetcher:
+    """
+    Handle PyPI Package URL (PURL) resolution and download URL retrieval.
+
+    Adapted from fetchcode
+    https://github.com/aboutcode-org/fetchcode/issues/190
+    """
+
+    purl_pattern = "pkg:pypi/.*"
+    base_url = "https://pypi.org/pypi"
+
+    @staticmethod
+    def fetch_json_response(url):
+        """Fetch a JSON response from the given URL and return the parsed JSON data."""
+        response = requests.get(url, timeout=DEFAULT_TIMEOUT)
+        if response.status_code != 200:
+            raise Exception(f"Failed to fetch {url}: {response.status_code} {response.reason}")
+
+        try:
+            return response.json()
+        except ValueError as e:
+            raise Exception(f"Failed to parse JSON from {url}: {str(e)}")
+
+    @classmethod
+    def get_package_data(cls, purl):
+        """Fetch package data from PyPI API."""
+        parsed_purl = PackageURL.from_string(purl)
+
+        if parsed_purl.version:
+            api_url = f"{cls.base_url}/{parsed_purl.name}/{parsed_purl.version}/json"
+        else:
+            api_url = f"{cls.base_url}/{parsed_purl.name}/json"
+
+        return cls.fetch_json_response(api_url)
+
+    @classmethod
+    def get_urls_info(cls, purl):
+        """Collect URL info dicts from PyPI API."""
+        data = cls.get_package_data(purl)
+        return data.get("urls", [])
+
+    @classmethod
+    def get_download_url(cls, purl, preferred_type="sdist"):
+        """
+        Get a single download URL from PyPI API.
+        If no version is specified in the PURL, fetches the latest version.
+        """
+        urls_info = cls.get_urls_info(purl)
+
+        if not urls_info:
+            return
+
+        for url_info in urls_info:
+            if url_info.get("packagetype") == preferred_type:
+                return url_info["url"]
+
+        return urls_info[0]["url"]
+
+    @classmethod
+    def get_all_download_urls(cls, purl):
+        """
+        Get all download URLs from PyPI API.
+        If no version is specified in the PURL, fetches the latest version.
+        """
+        urls_info = cls.get_urls_info(purl)
+        return [url_info["url"] for url_info in urls_info if "url" in url_info]
+
+
+def infer_download_url(purl):
+    """
+    Infer the download URL for a package from its Package URL (purl).
+
+    Attempts resolution via ``purl2url`` first. Falls back to package-type-specific
+    resolvers (which may make HTTP requests) when ``purl2url`` cannot resolve the URL.
+    """
+    if isinstance(purl, PackageURL):
+        purl_data = purl
+        purl_str = str(purl)
+    else:
+        purl_data = PackageURL.from_string(purl)
+        purl_str = purl
+
+    if download_url := purl2url.get_download_url(purl_str):
+        return download_url
+
+    # PyPI is not supported by ``purl2url``, it requires an API call to resolve download URLs.
+    if purl_data.type == "pypi":
+        with suppress(Exception):
+            return PyPIFetcher.get_download_url(purl_str, preferred_type="sdist")