refactor: create ArtifactCache class

src/poetry/installation/chef.py

@@ -17,15 +17,14 @@
 from poetry.core.utils.helpers import temporary_directory
 from pyproject_hooks import quiet_subprocess_runner  # type: ignore[import]
 
-from poetry.utils.cache import get_cache_directory_for_link
 from poetry.utils.env import ephemeral_environment
 
 
 if TYPE_CHECKING:
     from contextlib import AbstractContextManager
 
-    from poetry.config.config import Config
     from poetry.repositories import RepositoryPool
+    from poetry.utils.cache import ArtifactCache
     from poetry.utils.env import Env
 
 
@@ -81,10 +80,12 @@ def install(self, requirements: Collection[str]) -> None:
 
 
 class Chef:
-    def __init__(self, config: Config, env: Env, pool: RepositoryPool) -> None:
+    def __init__(
+        self, artifact_cache: ArtifactCache, env: Env, pool: RepositoryPool
+    ) -> None:
         self._env = env
         self._pool = pool
-        self._cache_dir = config.artifacts_cache_directory
+        self._artifact_cache = artifact_cache
 
     def prepare(
         self, archive: Path, output_dir: Path | None = None, *, editable: bool = False
@@ -174,8 +175,8 @@ def _prepare_sdist(self, archive: Path, destination: Path | None = None) -> Path
                     sdist_dir = archive_dir
 
             if destination is None:
-                destination = get_cache_directory_for_link(
-                    self._cache_dir, Link(archive.as_uri())
+                destination = self._artifact_cache.get_cache_directory_for_link(
+                    Link(archive.as_uri())
                 )
 
             destination.mkdir(parents=True, exist_ok=True)

src/poetry/installation/executor.py

@@ -26,8 +26,7 @@
 from poetry.installation.wheel_installer import WheelInstaller
 from poetry.utils._compat import decode
 from poetry.utils.authenticator import Authenticator
-from poetry.utils.cache import get_cache_directory_for_link
-from poetry.utils.cache import get_cached_archive_for_link
+from poetry.utils.cache import ArtifactCache
 from poetry.utils.env import EnvCommandError
 from poetry.utils.helpers import atomic_open
 from poetry.utils.helpers import get_file_hash
@@ -78,12 +77,12 @@ def __init__(
         else:
             self._max_workers = 1
 
+        self._artifact_cache = ArtifactCache(cache_dir=config.artifacts_cache_directory)
         self._authenticator = Authenticator(
             config, self._io, disable_cache=disable_cache, pool_size=self._max_workers
         )
-        self._chef = Chef(config, self._env, pool)
+        self._chef = Chef(self._artifact_cache, self._env, pool)
         self._chooser = Chooser(pool, self._env, config)
-        self._artifacts_cache_dir = config.artifacts_cache_directory
 
         self._executor = ThreadPoolExecutor(max_workers=self._max_workers)
         self._total_operations = 0
@@ -695,18 +694,18 @@ def _download(self, operation: Install | Update) -> Path:
     def _download_link(self, operation: Install | Update, link: Link) -> Path:
         package = operation.package
 
-        output_dir = get_cache_directory_for_link(self._artifacts_cache_dir, link)
+        output_dir = self._artifact_cache.get_cache_directory_for_link(link)
         # Try to get cached original package for the link provided
-        original_archive = get_cached_archive_for_link(
-            self._env, self._artifacts_cache_dir, link, strict=True
+        original_archive = self._artifact_cache.get_cached_archive_for_link(
+            link, strict=True
         )
         if original_archive is None:
             # No cached original distributions was found, so we download and prepare it
             try:
                 original_archive = self._download_archive(operation, link)
             except BaseException:
-                cache_directory = get_cache_directory_for_link(
-                    self._artifacts_cache_dir, link
+                cache_directory = self._artifact_cache.get_cache_directory_for_link(
+                    link
                 )
                 cached_file = cache_directory.joinpath(link.filename)
                 # We can't use unlink(missing_ok=True) because it's not available
@@ -718,8 +717,10 @@ def _download_link(self, operation: Install | Update, link: Link) -> Path:
 
         # Get potential higher prioritized cached archive, otherwise it will fall back
         # to the original archive.
-        archive = get_cached_archive_for_link(
-            self._env, self._artifacts_cache_dir, link, strict=False
+        archive = self._artifact_cache.get_cached_archive_for_link(
+            link,
+            strict=False,
+            env=self._env,
         )
         # 'archive' can at this point never be None. Since we previously downloaded
         # an archive, we now should have something cached that we can use here
@@ -785,8 +786,7 @@ def _download_archive(self, operation: Install | Update, link: Link) -> Path:
 
         done = 0
         archive = (
-            get_cache_directory_for_link(self._artifacts_cache_dir, link)
-            / link.filename
+            self._artifact_cache.get_cache_directory_for_link(link) / link.filename
         )
         archive.parent.mkdir(parents=True, exist_ok=True)
         with atomic_open(archive) as f:

src/poetry/utils/cache.py

@@ -208,70 +208,81 @@ def _deserialize(self, data_raw: bytes) -> CacheItem[T]:
         return CacheItem(data, expires)
 
 
-def get_cached_archives_for_link(cache_dir: Path, link: Link) -> list[Path]:
-    cache_dir = get_cache_directory_for_link(cache_dir, link)
-
-    archive_types = ["whl", "tar.gz", "tar.bz2", "bz2", "zip"]
-    paths = []
-    for archive_type in archive_types:
-        for archive in cache_dir.glob(f"*.{archive_type}"):
-            paths.append(Path(archive))
-
-    return paths
-
-
-def get_cached_archive_for_link(
-    env: Env, cache_dir: Path, link: Link, *, strict: bool
-) -> Path | None:
-    archives = get_cached_archives_for_link(cache_dir, link)
-    if not archives:
-        return None
-
-    candidates: list[tuple[float | None, Path]] = []
-    for archive in archives:
-        if strict:
-            # in strict mode return the original cached archive instead of the
-            # prioritized archive type.
-            if link.filename == archive.name:
-                return archive
-            continue
-        if archive.suffix != ".whl":
-            candidates.append((float("inf"), archive))
-            continue
-
-        try:
-            wheel = Wheel(archive.name)
-        except InvalidWheelName:
-            continue
-
-        if not wheel.is_supported_by_environment(env):
-            continue
-
-        candidates.append(
-            (wheel.get_minimum_supported_index(env.supported_tags), archive),
-        )
+class ArtifactCache:
+    def __init__(self, *, cache_dir: Path) -> None:
+        self._cache_dir = cache_dir
+
+    def _get_cached_archives_for_link(self, link: Link) -> list[Path]:
+        cache_dir = self.get_cache_directory_for_link(link)
+
+        archive_types = ["whl", "tar.gz", "tar.bz2", "bz2", "zip"]
+        paths = []
+        for archive_type in archive_types:
+            for archive in cache_dir.glob(f"*.{archive_type}"):
+                paths.append(Path(archive))
+
+        return paths
+
+    def get_cached_archive_for_link(
+        self,
+        link: Link,
+        *,
+        strict: bool,
+        env: Env | None = None,
+    ) -> Path | None:
+        assert strict or env is not None
+
+        archives = self._get_cached_archives_for_link(link)
+        if not archives:
+            return None
+
+        candidates: list[tuple[float | None, Path]] = []
+        for archive in archives:
+            if strict:
+                # in strict mode return the original cached archive instead of the
+                # prioritized archive type.
+                if link.filename == archive.name:
+                    return archive
+                continue
+
+            assert env is not None
+
+            if archive.suffix != ".whl":
+                candidates.append((float("inf"), archive))
+                continue
 
-    if not candidates:
-        return None
+            try:
+                wheel = Wheel(archive.name)
+            except InvalidWheelName:
+                continue
 
-    return min(candidates)[1]
+            if not wheel.is_supported_by_environment(env):
+                continue
+
+            candidates.append(
+                (wheel.get_minimum_supported_index(env.supported_tags), archive),
+            )
+
+        if not candidates:
+            return None
 
+        return min(candidates)[1]
 
-def get_cache_directory_for_link(cache_dir: Path, link: Link) -> Path:
-    key_parts = {"url": link.url_without_fragment}
+    def get_cache_directory_for_link(self, link: Link) -> Path:
+        key_parts = {"url": link.url_without_fragment}
 
-    if link.hash_name is not None and link.hash is not None:
-        key_parts[link.hash_name] = link.hash
+        if link.hash_name is not None and link.hash is not None:
+            key_parts[link.hash_name] = link.hash
 
-    if link.subdirectory_fragment:
-        key_parts["subdirectory"] = link.subdirectory_fragment
+        if link.subdirectory_fragment:
+            key_parts["subdirectory"] = link.subdirectory_fragment
 
-    key = hashlib.sha256(
-        json.dumps(
-            key_parts, sort_keys=True, separators=(",", ":"), ensure_ascii=True
-        ).encode("ascii")
-    ).hexdigest()
+        key = hashlib.sha256(
+            json.dumps(
+                key_parts, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+            ).encode("ascii")
+        ).hexdigest()
 
-    split_key = [key[:2], key[2:4], key[4:6], key[6:]]
+        split_key = [key[:2], key[2:4], key[4:6], key[6:]]
 
-    return cache_dir.joinpath(*split_key)
+        return self._cache_dir.joinpath(*split_key)

tests/installation/test_chef.py

@@ -14,7 +14,7 @@
 from poetry.factory import Factory
 from poetry.installation.chef import Chef
 from poetry.repositories import RepositoryPool
-from poetry.utils.cache import get_cache_directory_for_link
+from poetry.utils.cache import ArtifactCache
 from poetry.utils.env import EnvManager
 from tests.repositories.test_pypi_repository import MockRepository
 
@@ -39,25 +39,38 @@ def setup(mocker: MockerFixture, pool: RepositoryPool) -> None:
     mocker.patch.object(Factory, "create_pool", return_value=pool)
 
 
-def test_prepare_sdist(config: Config, config_cache_dir: Path) -> None:
-    chef = Chef(config, EnvManager.get_system_env(), Factory.create_pool(config))
+@pytest.fixture()
+def artifact_cache(config: Config) -> ArtifactCache:
+    return ArtifactCache(cache_dir=config.artifacts_cache_directory)
+
+
+def test_prepare_sdist(
+    config: Config, config_cache_dir: Path, artifact_cache: ArtifactCache
+) -> None:
+    chef = Chef(
+        artifact_cache, EnvManager.get_system_env(), Factory.create_pool(config)
+    )
 
     archive = (
         Path(__file__)
         .parent.parent.joinpath("fixtures/distributions/demo-0.1.0.tar.gz")
         .resolve()
     )
 
-    destination = get_cache_directory_for_link(chef._cache_dir, Link(archive.as_uri()))
+    destination = artifact_cache.get_cache_directory_for_link(Link(archive.as_uri()))
 
     wheel = chef.prepare(archive)
 
     assert wheel.parent == destination
     assert wheel.name == "demo-0.1.0-py3-none-any.whl"
 
 
-def test_prepare_directory(config: Config, config_cache_dir: Path):
-    chef = Chef(config, EnvManager.get_system_env(), Factory.create_pool(config))
+def test_prepare_directory(
+    config: Config, config_cache_dir: Path, artifact_cache: ArtifactCache
+):
+    chef = Chef(
+        artifact_cache, EnvManager.get_system_env(), Factory.create_pool(config)
+    )
 
     archive = Path(__file__).parent.parent.joinpath("fixtures/simple_project").resolve()
 
@@ -71,10 +84,10 @@ def test_prepare_directory(config: Config, config_cache_dir: Path):
 
 
 def test_prepare_directory_with_extensions(
-    config: Config, config_cache_dir: Path
+    config: Config, config_cache_dir: Path, artifact_cache: ArtifactCache
 ) -> None:
     env = EnvManager.get_system_env()
-    chef = Chef(config, env, Factory.create_pool(config))
+    chef = Chef(artifact_cache, env, Factory.create_pool(config))
 
     archive = (
         Path(__file__)
@@ -91,8 +104,12 @@ def test_prepare_directory_with_extensions(
     os.unlink(wheel)
 
 
-def test_prepare_directory_editable(config: Config, config_cache_dir: Path):
-    chef = Chef(config, EnvManager.get_system_env(), Factory.create_pool(config))
+def test_prepare_directory_editable(
+    config: Config, config_cache_dir: Path, artifact_cache: ArtifactCache
+):
+    chef = Chef(
+        artifact_cache, EnvManager.get_system_env(), Factory.create_pool(config)
+    )
 
     archive = Path(__file__).parent.parent.joinpath("fixtures/simple_project").resolve()
 

tests/installation/test_executor.py

@@ -43,7 +43,6 @@
 
     from poetry.config.config import Config
     from poetry.installation.operations.operation import Operation
-    from poetry.utils.env import Env
     from poetry.utils.env import VirtualEnv
     from tests.types import FixtureDirGetter
 
@@ -538,11 +537,11 @@ def test_executor_should_delete_incomplete_downloads(
         side_effect=Exception("Download error"),
     )
     mocker.patch(
-        "poetry.installation.executor.get_cached_archive_for_link",
+        "poetry.installation.executor.ArtifactCache.get_cached_archive_for_link",
         return_value=None,
     )
     mocker.patch(
-        "poetry.installation.executor.get_cache_directory_for_link",
+        "poetry.installation.executor.ArtifactCache.get_cache_directory_for_link",
         return_value=Path(tmp_dir),
     )
 
@@ -761,7 +760,7 @@ def test_executor_should_write_pep610_url_references_for_wheel_urls(
     if is_artifact_cached:
         link_cached = fixture_dir("distributions") / "demo-0.1.0-py2.py3-none-any.whl"
         mocker.patch(
-            "poetry.installation.executor.get_cached_archive_for_link",
+            "poetry.installation.executor.ArtifactCache.get_cached_archive_for_link",
             return_value=link_cached,
         )
     download_spy = mocker.spy(Executor, "_download_archive")
@@ -840,17 +839,15 @@ def test_executor_should_write_pep610_url_references_for_non_wheel_urls(
         cached_sdist = fixture_dir("distributions") / "demo-0.1.0.tar.gz"
         cached_wheel = fixture_dir("distributions") / "demo-0.1.0-py2.py3-none-any.whl"
 
-        def mock_get_cached_archive_for_link_func(
-            _: Env, __: Path, ___: Link, strict: bool
-        ):
+        def mock_get_cached_archive_for_link_func(_: Link, *, strict: bool, **__: Any):
             if is_wheel_cached and not strict:
                 return cached_wheel
             if is_sdist_cached:
                 return cached_sdist
             return None
 
         mocker.patch(
-            "poetry.installation.executor.get_cached_archive_for_link",
+            "poetry.installation.executor.ArtifactCache.get_cached_archive_for_link",
             side_effect=mock_get_cached_archive_for_link_func,
         )