"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "poetry/installation/chooser.py" between
poetry-1.1.15.tar.gz and poetry-1.2.0.tar.gz

About: Poetry is a tool for dependency management and packaging in Python.

chooser.py  (poetry-1.1.15):chooser.py  (poetry-1.2.0)
from __future__ import annotations
import logging
import re import re
from typing import List from typing import TYPE_CHECKING
from typing import Tuple from typing import Any
from packaging.tags import Tag from packaging.tags import Tag
from poetry.core.packages.package import Package from poetry.config.config import Config
from poetry.core.packages.utils.link import Link from poetry.config.config import PackageFilterPolicy
from poetry.repositories.pool import Pool
from poetry.utils.env import Env
from poetry.utils.patterns import wheel_file_re from poetry.utils.patterns import wheel_file_re
if TYPE_CHECKING:
from poetry.core.packages.package import Package
from poetry.core.packages.utils.link import Link
from poetry.core.semver.version import Version
from poetry.repositories.pool import Pool
from poetry.utils.env import Env
logger = logging.getLogger(__name__)
class InvalidWheelName(Exception): class InvalidWheelName(Exception):
pass pass
class Wheel(object): class Wheel:
def __init__(self, filename): # type: (str) -> None def __init__(self, filename: str) -> None:
wheel_info = wheel_file_re.match(filename) wheel_info = wheel_file_re.match(filename)
if not wheel_info: if not wheel_info:
raise InvalidWheelName("{} is not a valid wheel filename.".format(fi lename)) raise InvalidWheelName(f"{filename} is not a valid wheel filename.")
self.filename = filename self.filename = filename
self.name = wheel_info.group("name").replace("_", "-") self.name = wheel_info.group("name").replace("_", "-")
self.version = wheel_info.group("ver").replace("_", "-") self.version = wheel_info.group("ver").replace("_", "-")
self.build_tag = wheel_info.group("build") self.build_tag = wheel_info.group("build")
self.pyversions = wheel_info.group("pyver").split(".") self.pyversions = wheel_info.group("pyver").split(".")
self.abis = wheel_info.group("abi").split(".") self.abis = wheel_info.group("abi").split(".")
self.plats = wheel_info.group("plat").split(".") self.plats = wheel_info.group("plat").split(".")
self.tags = { self.tags = {
Tag(x, y, z) for x in self.pyversions for y in self.abis for z in se lf.plats Tag(x, y, z) for x in self.pyversions for y in self.abis for z in se lf.plats
} }
def get_minimum_supported_index(self, tags): def get_minimum_supported_index(self, tags: list[Tag]) -> int | None:
indexes = [tags.index(t) for t in self.tags if t in tags] indexes = [tags.index(t) for t in self.tags if t in tags]
return min(indexes) if indexes else None return min(indexes) if indexes else None
def is_supported_by_environment(self, env): def is_supported_by_environment(self, env: Env) -> bool:
return bool(set(env.supported_tags).intersection(self.tags)) return bool(set(env.supported_tags).intersection(self.tags))
class Chooser: class Chooser:
""" """
A Chooser chooses an appropriate release archive for packages. A Chooser chooses an appropriate release archive for packages.
""" """
def __init__(self, pool, env): # type: (Pool, Env) -> None def __init__(self, pool: Pool, env: Env, config: Config | None = None) -> No ne:
self._pool = pool self._pool = pool
self._env = env self._env = env
self._config = config or Config.create()
self._no_binary_policy: PackageFilterPolicy = PackageFilterPolicy(
self._config.get("installer.no-binary", [])
)
def choose_for(self, package): # type: (Package) -> Link def choose_for(self, package: Package) -> Link:
""" """
Return the url of the selected archive for a given package. Return the url of the selected archive for a given package.
""" """
links = [] links = []
for link in self._get_links(package): for link in self._get_links(package):
if link.is_wheel and not Wheel(link.filename).is_supported_by_enviro if link.is_wheel:
nment( if not self._no_binary_policy.allows(package.name):
self._env logger.debug(
): "Skipping wheel for %s as requested in no binary policy
continue for"
" package (%s)",
link.filename,
package.name,
)
continue
if not Wheel(link.filename).is_supported_by_environment(self._en
v):
logger.debug(
"Skipping wheel %s as this is not supported by the curre
nt"
" environment",
link.filename,
)
continue
if link.ext in {".egg", ".exe", ".msi", ".rpm", ".srpm"}: if link.ext in {".egg", ".exe", ".msi", ".rpm", ".srpm"}:
logger.debug("Skipping unsupported distribution %s", link.filena me)
continue continue
links.append(link) links.append(link)
if not links: if not links:
raise RuntimeError( raise RuntimeError(f"Unable to find installation candidates for {pac
"Unable to find installation candidates for {}".format(package) kage}")
)
# Get the best link # Get the best link
chosen = max(links, key=lambda link: self._sort_key(package, link)) chosen = max(links, key=lambda link: self._sort_key(package, link))
if not chosen:
raise RuntimeError(
"Unable to find installation candidates for {}".format(package)
)
return chosen return chosen
def _get_links(self, package): # type: (Package) -> List[Link] def _get_links(self, package: Package) -> list[Link]:
if not package.source_type: if package.source_type:
if not self._pool.has_repository("pypi"): assert package.source_reference is not None
repository = self._pool.repositories[0]
else:
repository = self._pool.repository("pypi")
else:
repository = self._pool.repository(package.source_reference) repository = self._pool.repository(package.source_reference)
elif not self._pool.has_repository("pypi"):
repository = self._pool.repositories[0]
else:
repository = self._pool.repository("pypi")
links = repository.find_links_for_package(package) links = repository.find_links_for_package(package)
hashes = [f["hash"] for f in package.files] hashes = [f["hash"] for f in package.files]
if not hashes: if not hashes:
return links return links
selected_links = [] selected_links = []
for link in links: for link in links:
if not link.hash: if not link.hash:
selected_links.append(link) selected_links.append(link)
continue continue
assert link.hash_name is not None
h = link.hash_name + ":" + link.hash h = link.hash_name + ":" + link.hash
if h not in hashes: if h not in hashes:
logger.debug(
"Skipping %s as %s checksum does not match expected value",
link.filename,
link.hash_name,
)
continue continue
selected_links.append(link) selected_links.append(link)
if links and not selected_links: if links and not selected_links:
raise RuntimeError( raise RuntimeError(
"Retrieved digest for link {}({}) not in poetry.lock metadata {} f"Retrieved digest for link {link.filename}({h}) not in poetry.l
".format( ock"
link.filename, h, hashes f" metadata {hashes}"
)
) )
return selected_links return selected_links
def _sort_key(self, package, link): # type: (Package, Link) -> Tuple def _sort_key(
self, package: Package, link: Link
) -> tuple[int, int, int, Version, tuple[Any, ...], int]:
""" """
Function to pass as the `key` argument to a call to sorted() to sort Function to pass as the `key` argument to a call to sorted() to sort
InstallationCandidates by preference. InstallationCandidates by preference.
Returns a tuple such that tuples sorting as greater using Python's Returns a tuple such that tuples sorting as greater using Python's
default comparison operator are more preferred. default comparison operator are more preferred.
The preference is as follows: The preference is as follows:
First and foremost, candidates with allowed (matching) hashes are First and foremost, candidates with allowed (matching) hashes are
always preferred over candidates without matching hashes. This is always preferred over candidates without matching hashes. This is
because e.g. if the only candidate with an allowed hash is yanked, because e.g. if the only candidate with an allowed hash is yanked,
we still want to use that candidate. we still want to use that candidate.
skipping to change at line 142 skipping to change at line 172
If not finding wheels, they are sorted by version only. If not finding wheels, they are sorted by version only.
If finding wheels, then the sort order is by version, then: If finding wheels, then the sort order is by version, then:
1. existing installs 1. existing installs
2. wheels ordered via Wheel.support_index_min(self._supported_tags) 2. wheels ordered via Wheel.support_index_min(self._supported_tags)
3. source archives 3. source archives
If prefer_binary was set, then all wheels are sorted above sources. If prefer_binary was set, then all wheels are sorted above sources.
Note: it was considered to embed this logic into the Link Note: it was considered to embed this logic into the Link
comparison operators, but then different sdist links comparison operators, but then different sdist links
with the same version, would have to be considered equal with the same version, would have to be considered equal
""" """
support_num = len(self._env.supported_tags) build_tag: tuple[Any, ...] = ()
build_tag = ()
binary_preference = 0 binary_preference = 0
if link.is_wheel: if link.is_wheel:
wheel = Wheel(link.filename) wheel = Wheel(link.filename)
if not wheel.is_supported_by_environment(self._env): if not wheel.is_supported_by_environment(self._env):
raise RuntimeError( raise RuntimeError(
"{} is not a supported wheel for this platform. It " f"{wheel.filename} is not a supported wheel for this platfor
"can't be sorted.".format(wheel.filename) m. It "
"can't be sorted."
) )
# TODO: Binary preference # TODO: Binary preference
pri = -(wheel.get_minimum_supported_index(self._env.supported_tags)) pri = -(wheel.get_minimum_supported_index(self._env.supported_tags) or 0)
if wheel.build_tag is not None: if wheel.build_tag is not None:
match = re.match(r"^(\d+)(.*)$", wheel.build_tag) match = re.match(r"^(\d+)(.*)$", wheel.build_tag)
if not match:
raise ValueError(f"Unable to parse build tag: {wheel.build_t
ag}")
build_tag_groups = match.groups() build_tag_groups = match.groups()
build_tag = (int(build_tag_groups[0]), build_tag_groups[1]) build_tag = (int(build_tag_groups[0]), build_tag_groups[1])
else: # sdist else: # sdist
support_num = len(self._env.supported_tags)
pri = -support_num pri = -support_num
has_allowed_hash = int(self._is_link_hash_allowed_for_package(link, pack age)) has_allowed_hash = int(self._is_link_hash_allowed_for_package(link, pack age))
# TODO: Proper yank value yank_value = int(not link.yanked)
yank_value = 0
return ( return (
has_allowed_hash, has_allowed_hash,
yank_value, yank_value,
binary_preference, binary_preference,
package.version, package.version,
build_tag, build_tag,
pri, pri,
) )
def _is_link_hash_allowed_for_package( def _is_link_hash_allowed_for_package(self, link: Link, package: Package) ->
self, link, package bool:
): # type: (Link, Package) -> bool
if not link.hash: if not link.hash:
return True return True
assert link.hash_name is not None
h = link.hash_name + ":" + link.hash h = link.hash_name + ":" + link.hash
return h in {f["hash"] for f in package.files} return h in {f["hash"] for f in package.files}
 End of changes. 29 change blocks. 
47 lines changed or deleted 83 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)