is a tool for dependency management and packaging in Python.
| chef.py (poetry-1.1.15) | : | chef.py (poetry-1.2.0) |
|---|
|
| | from __future__ import annotations |
| | |
| import hashlib | | import hashlib |
| import json | | import json |
| | |
|
| | from pathlib import Path |
| from typing import TYPE_CHECKING | | from typing import TYPE_CHECKING |
| | |
|
| from poetry.core.packages.utils.link import Link | | from poetry.installation.chooser import InvalidWheelName |
| from poetry.utils._compat import Path | | from poetry.installation.chooser import Wheel |
| | |
| from .chooser import InvalidWheelName | | |
| from .chooser import Wheel | | |
| | |
| if TYPE_CHECKING: | | if TYPE_CHECKING: |
|
| from typing import List | | from poetry.core.packages.utils.link import Link |
| from typing import Optional | | |
| | |
| from poetry.config.config import Config | | from poetry.config.config import Config |
| from poetry.utils.env import Env | | from poetry.utils.env import Env |
| | |
| class Chef: | | class Chef: |
|
| def __init__(self, config, env): # type: (Config, Env) -> None | | def __init__(self, config: Config, env: Env) -> None: |
| self._config = config | | |
| self._env = env | | self._env = env |
| self._cache_dir = ( | | self._cache_dir = ( |
| Path(config.get("cache-dir")).expanduser().joinpath("artifacts") | | Path(config.get("cache-dir")).expanduser().joinpath("artifacts") |
| ) | | ) |
| | |
|
| def prepare(self, archive): # type: (Path) -> Path | | def get_cached_archive_for_link(self, link: Link) -> Path | None: |
| return archive | | |
| | |
| def prepare_sdist(self, archive): # type: (Path) -> Path | | |
| return archive | | |
| | |
| def prepare_wheel(self, archive): # type: (Path) -> Path | | |
| return archive | | |
| | |
| def should_prepare(self, archive): # type: (Path) -> bool | | |
| return not self.is_wheel(archive) | | |
| | |
| def is_wheel(self, archive): # type: (Path) -> bool | | |
| return archive.suffix == ".whl" | | |
| | |
| def get_cached_archive_for_link(self, link): # type: (Link) -> Optional[Lin | | |
| k] | | |
| # If the archive is already a wheel, there is no need to cache it. | | |
| if link.is_wheel: | | |
| pass | | |
| | | |
| archives = self.get_cached_archives_for_link(link) | | archives = self.get_cached_archives_for_link(link) |
|
| | | |
| if not archives: | | if not archives: |
|
| return link | | return None |
| | |
|
| candidates = [] | | candidates: list[tuple[float | None, Path]] = [] |
| for archive in archives: | | for archive in archives: |
|
| if not archive.is_wheel: | | if archive.suffix != ".whl": |
| candidates.append((float("inf"), archive)) | | candidates.append((float("inf"), archive)) |
| continue | | continue |
| | |
| try: | | try: |
|
| wheel = Wheel(archive.filename) | | wheel = Wheel(archive.name) |
| except InvalidWheelName: | | except InvalidWheelName: |
| continue | | continue |
| | |
| if not wheel.is_supported_by_environment(self._env): | | if not wheel.is_supported_by_environment(self._env): |
| continue | | continue |
| | |
| candidates.append( | | candidates.append( |
| (wheel.get_minimum_supported_index(self._env.supported_tags), ar
chive), | | (wheel.get_minimum_supported_index(self._env.supported_tags), ar
chive), |
| ) | | ) |
| | |
| if not candidates: | | if not candidates: |
|
| return link | | return None |
| | |
| return min(candidates)[1] | | return min(candidates)[1] |
| | |
|
| def get_cached_archives_for_link(self, link): # type: (Link) -> List[Link] | | def get_cached_archives_for_link(self, link: Link) -> list[Path]: |
| cache_dir = self.get_cache_directory_for_link(link) | | cache_dir = self.get_cache_directory_for_link(link) |
| | |
| archive_types = ["whl", "tar.gz", "tar.bz2", "bz2", "zip"] | | archive_types = ["whl", "tar.gz", "tar.bz2", "bz2", "zip"] |
|
| links = [] | | paths = [] |
| for archive_type in archive_types: | | for archive_type in archive_types: |
|
| for archive in cache_dir.glob("*.{}".format(archive_type)): | | for archive in cache_dir.glob(f"*.{archive_type}"): |
| links.append(Link(archive.as_uri())) | | paths.append(Path(archive)) |
| | |
|
| return links | | return paths |
| | |
|
| def get_cache_directory_for_link(self, link): # type: (Link) -> Path | | def get_cache_directory_for_link(self, link: Link) -> Path: |
| key_parts = {"url": link.url_without_fragment} | | key_parts = {"url": link.url_without_fragment} |
| | |
| if link.hash_name is not None and link.hash is not None: | | if link.hash_name is not None and link.hash is not None: |
| key_parts[link.hash_name] = link.hash | | key_parts[link.hash_name] = link.hash |
| | |
| if link.subdirectory_fragment: | | if link.subdirectory_fragment: |
| key_parts["subdirectory"] = link.subdirectory_fragment | | key_parts["subdirectory"] = link.subdirectory_fragment |
| | |
| key_parts["interpreter_name"] = self._env.marker_env["interpreter_name"] | | key_parts["interpreter_name"] = self._env.marker_env["interpreter_name"] |
| key_parts["interpreter_version"] = "".join( | | key_parts["interpreter_version"] = "".join( |
| | | |
| End of changes. 17 change blocks. |
|---|
| 42 lines changed or deleted | | 19 lines changed or added |
|---|
"Fossies" - the Fresh Open Source Software Archive 