Polina/robot-control-system
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Загрузить файлы в «venv/Lib/site-packages/pip/_internal/resolution/resolvelib»

Browse Source
This commit is contained in:
Polina
2026-07-02 19:09:09 +00:00
parent 0c6a8edccb
commit 7a0187028e
4 changed files with 966 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

285
venv/Lib/site-packages/pip/_internal/resolution/resolvelib/provider.py Normal file
View File

@@ -0,0 +1,285 @@
from __future__ import annotations
import math
from collections.abc import Iterable, Iterator, Mapping, Sequence
from functools import cache
from typing import (
TYPE_CHECKING,
TypeVar,
)
from pip._vendor.resolvelib.providers import AbstractProvider
from pip._internal.req.req_install import InstallRequirement
from .base import Candidate, Constraint, Requirement
from .candidates import REQUIRES_PYTHON_IDENTIFIER
from .factory import Factory
from .requirements import ExplicitRequirement
if TYPE_CHECKING:
from pip._vendor.resolvelib.providers import Preference
from pip._vendor.resolvelib.resolvers import RequirementInformation
PreferenceInformation = RequirementInformation[Requirement, Candidate]
_ProviderBase = AbstractProvider[Requirement, Candidate, str]
else:
_ProviderBase = AbstractProvider
# Notes on the relationship between the provider, the factory, and the
# candidate and requirement classes.
#
# The provider is a direct implementation of the resolvelib class. Its role
# is to deliver the API that resolvelib expects.
#
# Rather than work with completely abstract "requirement" and "candidate"
# concepts as resolvelib does, pip has concrete classes implementing these two
# ideas. The API of Requirement and Candidate objects are defined in the base
# classes, but essentially map fairly directly to the equivalent provider
# methods. In particular, `find_matches` and `is_satisfied_by` are
# requirement methods, and `get_dependencies` is a candidate method.
#
# The factory is the interface to pip's internal mechanisms. It is stateless,
# and is created by the resolver and held as a property of the provider. It is
# responsible for creating Requirement and Candidate objects, and provides
# services to those objects (access to pip's finder and preparer).
D = TypeVar("D")
V = TypeVar("V")
def _get_with_identifier(
mapping: Mapping[str, V],
identifier: str,
default: D,
) -> D | V:
"""Get item from a package name lookup mapping with a resolver identifier.
This extra logic is needed when the target mapping is keyed by package
name, which cannot be directly looked up with an identifier (which may
contain requested extras). Additional logic is added to also look up a value
by "cleaning up" the extras from the identifier.
"""
if identifier in mapping:
return mapping[identifier]
# HACK: Theoretically we should check whether this identifier is a valid
# "NAME[EXTRAS]" format, and parse out the name part with packaging or
# some regular expression. But since pip's resolver only spits out three
# kinds of identifiers: normalized PEP 503 names, normalized names plus
# extras, and Requires-Python, we can cheat a bit here.
name, open_bracket, _ = identifier.partition("[")
if open_bracket and name in mapping:
return mapping[name]
return default
class PipProvider(_ProviderBase):
"""Pip's provider implementation for resolvelib.
:params constraints: A mapping of constraints specified by the user. Keys
are canonicalized project names.
:params ignore_dependencies: Whether the user specified ``--no-deps``.
:params upgrade_strategy: The user-specified upgrade strategy.
:params user_requested: A set of canonicalized package names that the user
supplied for pip to install/upgrade.
"""
def __init__(
self,
factory: Factory,
constraints: dict[str, Constraint],
ignore_dependencies: bool,
upgrade_strategy: str,
user_requested: dict[str, int],
) -> None:
self._factory = factory
self._constraints = constraints
self._ignore_dependencies = ignore_dependencies
self._upgrade_strategy = upgrade_strategy
self._user_requested = user_requested
@property
def constraints(self) -> dict[str, Constraint]:
"""Public view of user-specified constraints.
Exposes the provider's constraints mapping without encouraging
external callers to reach into private attributes.
"""
return self._constraints
def identify(self, requirement_or_candidate: Requirement | Candidate) -> str:
return requirement_or_candidate.name
def narrow_requirement_selection(
self,
identifiers: Iterable[str],
resolutions: Mapping[str, Candidate],
candidates: Mapping[str, Iterator[Candidate]],
information: Mapping[str, Iterator[PreferenceInformation]],
backtrack_causes: Sequence[PreferenceInformation],
) -> Iterable[str]:
"""Produce a subset of identifiers that should be considered before others.
Currently pip narrows the following selection:
* Requires-Python, if present is always returned by itself
* Backtrack causes are considered next because they can be identified
in linear time here, whereas because get_preference() is called
for each identifier, it would be quadratic to check for them there.
Further, the current backtrack causes likely need to be resolved
before other requirements as a resolution can't be found while
there is a conflict.
"""
backtrack_identifiers = set()
for info in backtrack_causes:
backtrack_identifiers.add(info.requirement.name)
if info.parent is not None:
backtrack_identifiers.add(info.parent.name)
current_backtrack_causes = []
for identifier in identifiers:
# Requires-Python has only one candidate and the check is basically
# free, so we always do it first to avoid needless work if it fails.
# This skips calling get_preference() for all other identifiers.
if identifier == REQUIRES_PYTHON_IDENTIFIER:
return [identifier]
# Check if this identifier is a backtrack cause
if identifier in backtrack_identifiers:
current_backtrack_causes.append(identifier)
continue
if current_backtrack_causes:
return current_backtrack_causes
return identifiers
def get_preference(
self,
identifier: str,
resolutions: Mapping[str, Candidate],
candidates: Mapping[str, Iterator[Candidate]],
information: Mapping[str, Iterable[PreferenceInformation]],
backtrack_causes: Sequence[PreferenceInformation],
) -> Preference:
"""Produce a sort key for given requirement based on preference.
The lower the return value is, the more preferred this group of
arguments is.
Currently pip considers the following in order:
* Any requirement that is "direct", e.g., points to an explicit URL.
* Any requirement that is "pinned", i.e., contains the operator ``===``
or ``==`` without a wildcard.
* Any requirement that imposes an upper version limit, i.e., contains the
operator ``<``, ``<=``, ``~=``, or ``==`` with a wildcard. Because
pip prioritizes the latest version, preferring explicit upper bounds
can rule out infeasible candidates sooner. This does not imply that
upper bounds are good practice; they can make dependency management
and resolution harder.
* Order user-specified requirements as they are specified, placing
other requirements afterward.
* Any "non-free" requirement, i.e., one that contains at least one
operator, such as ``>=`` or ``!=``.
* Alphabetical order for consistency (aids debuggability).
"""
try:
next(iter(information[identifier]))
except StopIteration:
# There is no information for this identifier, so there's no known
# candidates.
has_information = False
else:
has_information = True
if not has_information:
direct = False
ireqs: tuple[InstallRequirement | None, ...] = ()
else:
# Go through the information and for each requirement,
# check if it's explicit (e.g., a direct link) and get the
# InstallRequirement (the second element) from get_candidate_lookup()
directs, ireqs = zip(
*(
(isinstance(r, ExplicitRequirement), r.get_candidate_lookup()[1])
for r, _ in information[identifier]
)
)
direct = any(directs)
operators: list[tuple[str, str]] = [
(specifier.operator, specifier.version)
for specifier_set in (ireq.specifier for ireq in ireqs if ireq)
for specifier in specifier_set
]
pinned = any(((op[:2] == "==") and ("*" not in ver)) for op, ver in operators)
upper_bounded = any(
((op in ("<", "<=", "~=")) or (op == "==" and "*" in ver))
for op, ver in operators
)
unfree = bool(operators)
requested_order = self._user_requested.get(identifier, math.inf)
return (
not direct,
not pinned,
not upper_bounded,
requested_order,
not unfree,
identifier,
)
def find_matches(
self,
identifier: str,
requirements: Mapping[str, Iterator[Requirement]],
incompatibilities: Mapping[str, Iterator[Candidate]],
) -> Iterable[Candidate]:
def _eligible_for_upgrade(identifier: str) -> bool:
"""Are upgrades allowed for this project?
This checks the upgrade strategy, and whether the project was one
that the user specified in the command line, in order to decide
whether we should upgrade if there's a newer version available.
(Note that we don't need access to the `--upgrade` flag, because
an upgrade strategy of "to-satisfy-only" means that `--upgrade`
was not specified).
"""
if self._upgrade_strategy == "eager":
return True
elif self._upgrade_strategy == "only-if-needed":
user_order = _get_with_identifier(
self._user_requested,
identifier,
default=None,
)
return user_order is not None
return False
constraint = _get_with_identifier(
self._constraints,
identifier,
default=Constraint.empty(),
)
return self._factory.find_candidates(
identifier=identifier,
requirements=requirements,
constraint=constraint,
prefers_installed=(not _eligible_for_upgrade(identifier)),
incompatibilities=incompatibilities,
is_satisfied_by=self.is_satisfied_by,
)
@staticmethod
@cache
def is_satisfied_by(requirement: Requirement, candidate: Candidate) -> bool:
return requirement.is_satisfied_by(candidate)
def get_dependencies(self, candidate: Candidate) -> Iterable[Requirement]:
with_requires = not self._ignore_dependencies
# iter_dependencies() can perform nontrivial work so delay until needed.
return (r for r in candidate.iter_dependencies(with_requires) if r is not None)

98
venv/Lib/site-packages/pip/_internal/resolution/resolvelib/reporter.py Normal file
View File

@@ -0,0 +1,98 @@
from __future__ import annotations
from collections import defaultdict
from collections.abc import Mapping
from logging import getLogger
from typing import Any
from pip._vendor.resolvelib.reporters import BaseReporter
from .base import Candidate, Constraint, Requirement
logger = getLogger(__name__)
class PipReporter(BaseReporter[Requirement, Candidate, str]):
def __init__(self, constraints: Mapping[str, Constraint] | None = None) -> None:
self.reject_count_by_package: defaultdict[str, int] = defaultdict(int)
self._constraints = constraints or {}
self._messages_at_reject_count = {
1: (
"pip is looking at multiple versions of {package_name} to "
"determine which version is compatible with other "
"requirements. This could take a while."
),
8: (
"pip is still looking at multiple versions of {package_name} to "
"determine which version is compatible with other "
"requirements. This could take a while."
),
13: (
"This is taking longer than usual. You might need to provide "
"the dependency resolver with stricter constraints to reduce "
"runtime. See https://pip.pypa.io/warnings/backtracking for "
"guidance. If you want to abort this run, press Ctrl + C."
),
}
def rejecting_candidate(self, criterion: Any, candidate: Candidate) -> None:
"""Report a candidate being rejected.
Logs both the rejection count message (if applicable) and details about
the requirements and constraints that caused the rejection.
"""
self.reject_count_by_package[candidate.name] += 1
count = self.reject_count_by_package[candidate.name]
if count in self._messages_at_reject_count:
message = self._messages_at_reject_count[count]
logger.info("INFO: %s", message.format(package_name=candidate.name))
msg = "Will try a different candidate, due to conflict:"
for req_info in criterion.information:
req, parent = req_info.requirement, req_info.parent
msg += "\n "
if parent:
msg += f"{parent.name} {parent.version} depends on "
else:
msg += "The user requested "
msg += req.format_for_error()
# Add any relevant constraints
if self._constraints:
name = candidate.name
constraint = self._constraints.get(name)
if constraint and constraint.specifier:
constraint_text = f"{name}{constraint.specifier}"
msg += f"\n The user requested (constraint) {constraint_text}"
logger.debug(msg)
class PipDebuggingReporter(BaseReporter[Requirement, Candidate, str]):
"""A reporter that does an info log for every event it sees."""
def starting(self) -> None:
logger.info("Reporter.starting()")
def starting_round(self, index: int) -> None:
logger.info("Reporter.starting_round(%r)", index)
def ending_round(self, index: int, state: Any) -> None:
logger.info("Reporter.ending_round(%r, state)", index)
logger.debug("Reporter.ending_round(%r, %r)", index, state)
def ending(self, state: Any) -> None:
logger.info("Reporter.ending(%r)", state)
def adding_requirement(
self, requirement: Requirement, parent: Candidate | None
) -> None:
logger.info("Reporter.adding_requirement(%r, %r)", requirement, parent)
def rejecting_candidate(self, criterion: Any, candidate: Candidate) -> None:
logger.info("Reporter.rejecting_candidate(%r, %r)", criterion, candidate)
def pinning(self, candidate: Candidate) -> None:
logger.info("Reporter.pinning(%r)", candidate)

251
venv/Lib/site-packages/pip/_internal/resolution/resolvelib/requirements.py Normal file
View File

@@ -0,0 +1,251 @@
from __future__ import annotations
from typing import Any
from pip._vendor.packaging.specifiers import SpecifierSet
from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
from pip._internal.req.constructors import install_req_drop_extras
from pip._internal.req.req_install import InstallRequirement
from .base import Candidate, CandidateLookup, Requirement, format_name
class ExplicitRequirement(Requirement):
def __init__(self, candidate: Candidate) -> None:
self.candidate = candidate
def __str__(self) -> str:
return str(self.candidate)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.candidate!r})"
def __hash__(self) -> int:
return hash(self.candidate)
def __eq__(self, other: Any) -> bool:
if not isinstance(other, ExplicitRequirement):
return False
return self.candidate == other.candidate
@property
def project_name(self) -> NormalizedName:
# No need to canonicalize - the candidate did this
return self.candidate.project_name
@property
def name(self) -> str:
# No need to canonicalize - the candidate did this
return self.candidate.name
def format_for_error(self) -> str:
return self.candidate.format_for_error()
def get_candidate_lookup(self) -> CandidateLookup:
return self.candidate, None
def is_satisfied_by(self, candidate: Candidate) -> bool:
return candidate == self.candidate
class SpecifierRequirement(Requirement):
def __init__(self, ireq: InstallRequirement) -> None:
assert ireq.link is None, "This is a link, not a specifier"
self._ireq = ireq
self._equal_cache: str | None = None
self._hash: int | None = None
self._extras = frozenset(canonicalize_name(e) for e in self._ireq.extras)
@property
def _equal(self) -> str:
if self._equal_cache is not None:
return self._equal_cache
self._equal_cache = str(self._ireq)
return self._equal_cache
def __str__(self) -> str:
return str(self._ireq.req)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({str(self._ireq.req)!r})"
def __eq__(self, other: object) -> bool:
if not isinstance(other, SpecifierRequirement):
return NotImplemented
return self._equal == other._equal
def __hash__(self) -> int:
if self._hash is not None:
return self._hash
self._hash = hash(self._equal)
return self._hash
@property
def project_name(self) -> NormalizedName:
assert self._ireq.req, "Specifier-backed ireq is always PEP 508"
return canonicalize_name(self._ireq.req.name)
@property
def name(self) -> str:
return format_name(self.project_name, self._extras)
def format_for_error(self) -> str:
# Convert comma-separated specifiers into "A, B, ..., F and G"
# This makes the specifier a bit more "human readable", without
# risking a change in meaning. (Hopefully! Not all edge cases have
# been checked)
parts = [s.strip() for s in str(self).split(",")]
if len(parts) == 0:
return ""
elif len(parts) == 1:
return parts[0]
return ", ".join(parts[:-1]) + " and " + parts[-1]
def get_candidate_lookup(self) -> CandidateLookup:
return None, self._ireq
def is_satisfied_by(self, candidate: Candidate) -> bool:
assert candidate.name == self.name, (
f"Internal issue: Candidate is not for this requirement "
f"{candidate.name} vs {self.name}"
)
# We can safely always allow prereleases here since PackageFinder
# already implements the prerelease logic, and would have filtered out
# prerelease candidates if the user does not expect them.
assert self._ireq.req, "Specifier-backed ireq is always PEP 508"
spec = self._ireq.req.specifier
return spec.contains(candidate.version, prereleases=True)
class SpecifierWithoutExtrasRequirement(SpecifierRequirement):
"""
Requirement backed by an install requirement on a base package.
Trims extras from its install requirement if there are any.
"""
def __init__(self, ireq: InstallRequirement) -> None:
assert ireq.link is None, "This is a link, not a specifier"
self._ireq = install_req_drop_extras(ireq)
self._equal_cache: str | None = None
self._hash: int | None = None
self._extras = frozenset(canonicalize_name(e) for e in self._ireq.extras)
@property
def _equal(self) -> str:
if self._equal_cache is not None:
return self._equal_cache
self._equal_cache = str(self._ireq)
return self._equal_cache
def __eq__(self, other: object) -> bool:
if not isinstance(other, SpecifierWithoutExtrasRequirement):
return NotImplemented
return self._equal == other._equal
def __hash__(self) -> int:
if self._hash is not None:
return self._hash
self._hash = hash(self._equal)
return self._hash
class RequiresPythonRequirement(Requirement):
"""A requirement representing Requires-Python metadata."""
def __init__(self, specifier: SpecifierSet, match: Candidate) -> None:
self.specifier = specifier
self._specifier_string = str(specifier) # for faster __eq__
self._hash: int | None = None
self._candidate = match
# Pre-compute candidate lookup to avoid repeated specifier checks
if specifier.contains(match.version, prereleases=True):
self._candidate_lookup: CandidateLookup = (match, None)
else:
self._candidate_lookup = (None, None)
def __str__(self) -> str:
return f"Python {self.specifier}"
def __repr__(self) -> str:
return f"{self.__class__.__name__}({str(self.specifier)!r})"
def __hash__(self) -> int:
if self._hash is not None:
return self._hash
self._hash = hash((self._specifier_string, self._candidate))
return self._hash
def __eq__(self, other: Any) -> bool:
if not isinstance(other, RequiresPythonRequirement):
return False
return (
self._specifier_string == other._specifier_string
and self._candidate == other._candidate
)
@property
def project_name(self) -> NormalizedName:
return self._candidate.project_name
@property
def name(self) -> str:
return self._candidate.name
def format_for_error(self) -> str:
return str(self)
def get_candidate_lookup(self) -> CandidateLookup:
return self._candidate_lookup
def is_satisfied_by(self, candidate: Candidate) -> bool:
assert candidate.name == self._candidate.name, "Not Python candidate"
# We can safely always allow prereleases here since PackageFinder
# already implements the prerelease logic, and would have filtered out
# prerelease candidates if the user does not expect them.
return self.specifier.contains(candidate.version, prereleases=True)
class UnsatisfiableRequirement(Requirement):
"""A requirement that cannot be satisfied."""
def __init__(self, name: NormalizedName) -> None:
self._name = name
def __str__(self) -> str:
return f"{self._name} (unavailable)"
def __repr__(self) -> str:
return f"{self.__class__.__name__}({str(self._name)!r})"
def __eq__(self, other: object) -> bool:
if not isinstance(other, UnsatisfiableRequirement):
return NotImplemented
return self._name == other._name
def __hash__(self) -> int:
return hash(self._name)
@property
def project_name(self) -> NormalizedName:
return self._name
@property
def name(self) -> str:
return self._name
def format_for_error(self) -> str:
return str(self)
def get_candidate_lookup(self) -> CandidateLookup:
return None, None
def is_satisfied_by(self, candidate: Candidate) -> bool:
return False

332
venv/Lib/site-packages/pip/_internal/resolution/resolvelib/resolver.py Normal file
View File

@@ -0,0 +1,332 @@
from __future__ import annotations
import contextlib
import functools
import logging
import os
from typing import TYPE_CHECKING, cast
from pip._vendor.packaging.utils import canonicalize_name
from pip._vendor.resolvelib import BaseReporter, ResolutionImpossible, ResolutionTooDeep
from pip._vendor.resolvelib import Resolver as RLResolver
from pip._vendor.resolvelib.structs import DirectedGraph
from pip._internal.cache import WheelCache
from pip._internal.exceptions import ResolutionTooDeepError
from pip._internal.index.package_finder import PackageFinder
from pip._internal.operations.prepare import RequirementPreparer
from pip._internal.req.constructors import install_req_extend_extras
from pip._internal.req.req_install import InstallRequirement
from pip._internal.req.req_set import RequirementSet
from pip._internal.resolution.base import BaseResolver, InstallRequirementProvider
from pip._internal.resolution.resolvelib.provider import PipProvider
from pip._internal.resolution.resolvelib.reporter import (
PipDebuggingReporter,
PipReporter,
)
from pip._internal.utils.packaging import get_requirement
from .base import Candidate, Requirement
from .factory import Factory
if TYPE_CHECKING:
from pip._vendor.resolvelib.resolvers import Result as RLResult
Result = RLResult[Requirement, Candidate, str]
logger = logging.getLogger(__name__)
class Resolver(BaseResolver):
_allowed_strategies = {"eager", "only-if-needed", "to-satisfy-only"}
def __init__(
self,
preparer: RequirementPreparer,
finder: PackageFinder,
wheel_cache: WheelCache | None,
make_install_req: InstallRequirementProvider,
use_user_site: bool,
ignore_dependencies: bool,
ignore_installed: bool,
ignore_requires_python: bool,
force_reinstall: bool,
upgrade_strategy: str,
py_version_info: tuple[int, ...] | None = None,
):
super().__init__()
assert upgrade_strategy in self._allowed_strategies
self.factory = Factory(
finder=finder,
preparer=preparer,
make_install_req=make_install_req,
wheel_cache=wheel_cache,
use_user_site=use_user_site,
force_reinstall=force_reinstall,
ignore_installed=ignore_installed,
ignore_requires_python=ignore_requires_python,
py_version_info=py_version_info,
)
self.ignore_dependencies = ignore_dependencies
self.upgrade_strategy = upgrade_strategy
self._result: Result | None = None
def resolve(
self, root_reqs: list[InstallRequirement], check_supported_wheels: bool
) -> RequirementSet:
collected = self.factory.collect_root_requirements(root_reqs)
provider = PipProvider(
factory=self.factory,
constraints=collected.constraints,
ignore_dependencies=self.ignore_dependencies,
upgrade_strategy=self.upgrade_strategy,
user_requested=collected.user_requested,
)
if "PIP_RESOLVER_DEBUG" in os.environ:
reporter: BaseReporter[Requirement, Candidate, str] = PipDebuggingReporter()
else:
reporter = PipReporter(constraints=provider.constraints)
resolver: RLResolver[Requirement, Candidate, str] = RLResolver(
provider,
reporter,
)
try:
limit_how_complex_resolution_can_be = 200000
result = self._result = resolver.resolve(
collected.requirements, max_rounds=limit_how_complex_resolution_can_be
)
except ResolutionImpossible as e:
error = self.factory.get_installation_error(
cast("ResolutionImpossible[Requirement, Candidate]", e),
collected.constraints,
)
raise error from e
except ResolutionTooDeep:
raise ResolutionTooDeepError from None
req_set = RequirementSet(check_supported_wheels=check_supported_wheels)
# process candidates with extras last to ensure their base equivalent is
# already in the req_set if appropriate.
# Python's sort is stable so using a binary key function keeps relative order
# within both subsets.
for candidate in sorted(
result.mapping.values(), key=lambda c: c.name != c.project_name
):
ireq = candidate.get_install_requirement()
if ireq is None:
if candidate.name != candidate.project_name:
# extend existing req's extras
with contextlib.suppress(KeyError):
req = req_set.get_requirement(candidate.project_name)
req_set.add_named_requirement(
install_req_extend_extras(
req, get_requirement(candidate.name).extras
)
)
continue
# Check if there is already an installation under the same name,
# and set a flag for later stages to uninstall it, if needed.
installed_dist = self.factory.get_dist_to_uninstall(candidate)
if installed_dist is None:
# There is no existing installation -- nothing to uninstall.
ireq.should_reinstall = False
elif self.factory.force_reinstall:
# The --force-reinstall flag is set -- reinstall.
ireq.should_reinstall = True
elif installed_dist.version != candidate.version:
# The installation is different in version -- reinstall.
ireq.should_reinstall = True
elif candidate.is_editable or installed_dist.editable:
# The incoming distribution is editable, or different in
# editable-ness to installation -- reinstall.
ireq.should_reinstall = True
elif candidate.source_link and candidate.source_link.is_file:
# The incoming distribution is under file://
if candidate.source_link.is_wheel:
# is a local wheel -- do nothing.
logger.info(
"%s is already installed with the same version as the "
"provided wheel. Use --force-reinstall to force an "
"installation of the wheel.",
ireq.name,
)
continue
# is a local sdist or path -- reinstall
ireq.should_reinstall = True
else:
continue
link = candidate.source_link
if link and link.is_yanked:
# The reason can contain non-ASCII characters, Unicode
# is required for Python 2.
msg = (
"The candidate selected for download or install is a "
"yanked version: {name!r} candidate (version {version} "
"at {link})\nReason for being yanked: {reason}"
).format(
name=candidate.name,
version=candidate.version,
link=link,
reason=link.yanked_reason or "<none given>",
)
logger.warning(msg)
req_set.add_named_requirement(ireq)
return req_set
def get_installation_order(
self, req_set: RequirementSet
) -> list[InstallRequirement]:
"""Get order for installation of requirements in RequirementSet.
The returned list contains a requirement before another that depends on
it. This helps ensure that the environment is kept consistent as they
get installed one-by-one.
The current implementation creates a topological ordering of the
dependency graph, giving more weight to packages with less
or no dependencies, while breaking any cycles in the graph at
arbitrary points. We make no guarantees about where the cycle
would be broken, other than it *would* be broken.
"""
assert self._result is not None, "must call resolve() first"
if not req_set.requirements:
# Nothing is left to install, so we do not need an order.
return []
graph = self._result.graph
weights = get_topological_weights(graph, set(req_set.requirements.keys()))
sorted_items = sorted(
req_set.requirements.items(),
key=functools.partial(_req_set_item_sorter, weights=weights),
reverse=True,
)
return [ireq for _, ireq in sorted_items]
def get_topological_weights(
graph: DirectedGraph[str | None], requirement_keys: set[str]
) -> dict[str | None, int]:
"""Assign weights to each node based on how "deep" they are.
This implementation may change at any point in the future without prior
notice.
We first simplify the dependency graph by pruning any leaves and giving them
the highest weight: a package without any dependencies should be installed
first. This is done again and again in the same way, giving ever less weight
to the newly found leaves. The loop stops when no leaves are left: all
remaining packages have at least one dependency left in the graph.
Then we continue with the remaining graph, by taking the length for the
longest path to any node from root, ignoring any paths that contain a single
node twice (i.e. cycles). This is done through a depth-first search through
the graph, while keeping track of the path to the node.
Cycles in the graph result would result in node being revisited while also
being on its own path. In this case, take no action. This helps ensure we
don't get stuck in a cycle.
When assigning weight, the longer path (i.e. larger length) is preferred.
We are only interested in the weights of packages that are in the
requirement_keys.
"""
path: set[str | None] = set()
weights: dict[str | None, list[int]] = {}
def visit(node: str | None) -> None:
if node in path:
# We hit a cycle, so we'll break it here.
return
# The walk is exponential and for pathologically connected graphs (which
# are the ones most likely to contain cycles in the first place) it can
# take until the heat-death of the universe. To counter this we limit
# the number of attempts to visit (i.e. traverse through) any given
# node. We choose a value here which gives decent enough coverage for
# fairly well behaved graphs, and still limits the walk complexity to be
# linear in nature.
cur_weights = weights.get(node, [])
if len(cur_weights) >= 5:
return
# Time to visit the children!
path.add(node)
for child in graph.iter_children(node):
visit(child)
path.remove(node)
if node not in requirement_keys:
return
cur_weights.append(len(path))
weights[node] = cur_weights
# Simplify the graph, pruning leaves that have no dependencies. This is
# needed for large graphs (say over 200 packages) because the `visit`
# function is slower for large/densely connected graphs, taking minutes.
# See https://github.com/pypa/pip/issues/10557
# We repeat the pruning step until we have no more leaves to remove.
while True:
leaves = set()
for key in graph:
if key is None:
continue
for _child in graph.iter_children(key):
# This means we have at least one child
break
else:
# No child.
leaves.add(key)
if not leaves:
# We are done simplifying.
break
# Calculate the weight for the leaves.
weight = len(graph) - 1
for leaf in leaves:
if leaf not in requirement_keys:
continue
weights[leaf] = [weight]
# Remove the leaves from the graph, making it simpler.
for leaf in leaves:
graph.remove(leaf)
# Visit the remaining graph, this will only have nodes to handle if the
# graph had a cycle in it, which the pruning step above could not handle.
# `None` is guaranteed to be the root node by resolvelib.
visit(None)
# Sanity check: all requirement keys should be in the weights,
# and no other keys should be in the weights.
difference = set(weights.keys()).difference(requirement_keys)
assert not difference, difference
# Now give back all the weights, choosing the largest ones from what we
# accumulated.
return {node: max(wgts) for (node, wgts) in weights.items()}
def _req_set_item_sorter(
item: tuple[str, InstallRequirement],
weights: dict[str | None, int],
) -> tuple[int, str]:
"""Key function used to sort install requirements for installation.
Based on the "weight" mapping calculated in ``get_installation_order()``.
The canonical package name is returned as the second member as a tie-
breaker to ensure the result is predictable, which is useful in tests.
"""
name = canonicalize_name(item[0])
return weights[name], name