Spaces:
Sleeping
Sleeping
File size: 9,914 Bytes
69d9940 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 |
import collections
import math
from typing import (
TYPE_CHECKING,
Dict,
Iterable,
Iterator,
Mapping,
Sequence,
TypeVar,
Union,
)
from pip._vendor.resolvelib.providers import AbstractProvider
from .base import Candidate, Constraint, Requirement
from .candidates import REQUIRES_PYTHON_IDENTIFIER
from .factory import Factory
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,
) -> Union[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
self._known_depths: Dict[str, float] = collections.defaultdict(lambda: math.inf)
def identify(self, requirement_or_candidate: Union[Requirement, Candidate]) -> str:
return requirement_or_candidate.name
def get_preference( # type: ignore
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:
* Prefer if any of the known requirements is "direct", e.g. points to an
explicit URL.
* If equal, prefer if any requirement is "pinned", i.e. contains
operator ``===`` or ``==``.
* If equal, calculate an approximate "depth" and resolve requirements
closer to the user-specified requirements first.
* Order user-specified requirements by the order they are specified.
* If equal, prefers "non-free" requirements, i.e. contains at least one
operator, such as ``>=`` or ``<``.
* If equal, order alphabetically for consistency (helps debuggability).
"""
lookups = (r.get_candidate_lookup() for r, _ in information[identifier])
candidate, ireqs = zip(*lookups)
operators = [
specifier.operator
for specifier_set in (ireq.specifier for ireq in ireqs if ireq)
for specifier in specifier_set
]
direct = candidate is not None
pinned = any(op[:2] == "==" for op in operators)
unfree = bool(operators)
try:
requested_order: Union[int, float] = self._user_requested[identifier]
except KeyError:
requested_order = math.inf
parent_depths = (
self._known_depths[parent.name] if parent is not None else 0.0
for _, parent in information[identifier]
)
inferred_depth = min(d for d in parent_depths) + 1.0
else:
inferred_depth = 1.0
self._known_depths[identifier] = inferred_depth
requested_order = self._user_requested.get(identifier, math.inf)
# 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.
requires_python = identifier == REQUIRES_PYTHON_IDENTIFIER
# HACK: Setuptools have a very long and solid backward compatibility
# track record, and extremely few projects would request a narrow,
# non-recent version range of it since that would break a lot things.
# (Most projects specify it only to request for an installer feature,
# which does not work, but that's another topic.) Intentionally
# delaying Setuptools helps reduce branches the resolver has to check.
# This serves as a temporary fix for issues like "apache-airflow[all]"
# while we work on "proper" branch pruning techniques.
delay_this = identifier == "setuptools"
# Prefer the causes of backtracking on the assumption that the problem
# resolving the dependency tree is related to the failures that caused
# the backtracking
backtrack_cause = self.is_backtrack_cause(identifier, backtrack_causes)
return (
not requires_python,
delay_this,
not direct,
not pinned,
not backtrack_cause,
inferred_depth,
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,
)
def is_satisfied_by(self, requirement: Requirement, candidate: Candidate) -> bool:
return requirement.is_satisfied_by(candidate)
def get_dependencies(self, candidate: Candidate) -> Sequence[Requirement]:
with_requires = not self._ignore_dependencies
return [r for r in candidate.iter_dependencies(with_requires) if r is not None]
@staticmethod
def is_backtrack_cause(
identifier: str, backtrack_causes: Sequence["PreferenceInformation"]
) -> bool:
for backtrack_cause in backtrack_causes:
if identifier == backtrack_cause.requirement.name:
return True
if backtrack_cause.parent and identifier == backtrack_cause.parent.name:
return True
return False
|