perf: use cached file for wheel inspection (#7916)

src/poetry/repositories/http_repository.py

@@ -2,14 +2,13 @@
 
 import functools
 import hashlib
-import os
-import urllib
-import urllib.parse
 
 from collections import defaultdict
+from contextlib import contextmanager
 from pathlib import Path
 from typing import TYPE_CHECKING
 from typing import Any
+from typing import Iterator
 
 import requests
 
@@ -75,34 +74,28 @@ def authenticated_url(self) -> str:
     def _download(self, url: str, dest: Path) -> None:
         return download_file(url, dest, session=self.session)
 
+    @contextmanager
+    def _cached_or_downloaded_file(self, link: Link) -> Iterator[Path]:
+        filepath = self._authenticator.get_cached_file_for_url(link.url)
+        if filepath:
+            yield filepath
+        else:
+            self._log(f"Downloading: {link.url}", level="debug")
+            with temporary_directory() as temp_dir:
+                filepath = Path(temp_dir) / link.filename
+                self._download(link.url, filepath)
+                yield filepath
+
     def _get_info_from_wheel(self, url: str) -> PackageInfo:
         from poetry.inspection.info import PackageInfo
 
-        wheel_name = urllib.parse.urlparse(url).path.rsplit("/")[-1]
-        self._log(f"Downloading wheel: {wheel_name}", level="debug")
-
-        filename = os.path.basename(wheel_name)
-
-        with temporary_directory() as temp_dir:
-            filepath = Path(temp_dir) / filename
-            self._download(url, filepath)
-
+        with self._cached_or_downloaded_file(Link(url)) as filepath:
             return PackageInfo.from_wheel(filepath)
 
     def _get_info_from_sdist(self, url: str) -> PackageInfo:
         from poetry.inspection.info import PackageInfo
 
-        sdist_name = urllib.parse.urlparse(url).path
-        sdist_name_log = sdist_name.rsplit("/")[-1]
-
-        self._log(f"Downloading sdist: {sdist_name_log}", level="debug")
-
-        filename = os.path.basename(sdist_name)
-
-        with temporary_directory() as temp_dir:
-            filepath = Path(temp_dir) / filename
-            self._download(url, filepath)
-
+        with self._cached_or_downloaded_file(Link(url)) as filepath:
             return PackageInfo.from_sdist(filepath)
 
     def _get_info_from_urls(self, urls: dict[str, list[str]]) -> PackageInfo:
@@ -237,10 +230,7 @@ def _links_to_data(self, links: list[Link], data: PackageInfo) -> dict[str, Any]
                 and link.hash_name not in ("sha256", "sha384", "sha512")
                 and hasattr(hashlib, link.hash_name)
             ):
-                with temporary_directory() as temp_dir:
-                    filepath = Path(temp_dir) / link.filename
-                    self._download(link.url, filepath)
-
+                with self._cached_or_downloaded_file(link) as filepath:
                     known_hash = (
                         getattr(hashlib, link.hash_name)() if link.hash_name else None
                     )

src/poetry/utils/authenticator.py

@@ -19,6 +19,7 @@
 
 from cachecontrol import CacheControlAdapter
 from cachecontrol.caches import FileCache
+from cachecontrol.caches.file_cache import url_to_file_path
 from filelock import FileLock
 
 from poetry.config.config import Config
@@ -463,6 +464,13 @@ def _get_certs_for_url(self, url: str) -> RepositoryCertificateConfig:
             return selected.certs(config=self._config)
         return RepositoryCertificateConfig()
 
+    def get_cached_file_for_url(self, url: str) -> Path | None:
+        if self._cache_control is None:
+            return None
+
+        path = Path(url_to_file_path(url, self._cache_control))
+        return path if path.exists() else None
+
 
 _authenticator: Authenticator | None = None
 

tests/utils/test_authenticator.py

@@ -627,6 +627,22 @@ def test_authenticator_git_repositories(
     assert not three.password
 
 
+def test_authenticator_get_cached_file_for_url__cache_miss(config: Config) -> None:
+    authenticator = Authenticator(config, NullIO())
+    assert (
+        authenticator.get_cached_file_for_url("https://foo.bar/cache/miss.whl") is None
+    )
+
+
+def test_authenticator_get_cached_file_for_url__cache_hit(config: Config) -> None:
+    authenticator = Authenticator(config, NullIO())
+    url = "https://foo.bar/files/foo-0.1.0.tar.gz"
+
+    authenticator._cache_control.set(url, b"hello")
+
+    assert authenticator.get_cached_file_for_url(url)
+
+
 @pytest.mark.parametrize(
     ("ca_cert", "client_cert", "result"),
     [