import contextlib
import functools
import logging
from typing import (
TYPE_CHECKING,
Dict,
FrozenSet,
Iterable,
Iterator,
List,
Mapping,
NamedTuple,
Optional,
Sequence,
Set,
Tuple,
TypeVar,
cast,
)
from pip._vendor.packaging.requirements import InvalidRequirement
from pip._vendor.packaging.specifiers import SpecifierSet
from pip._vendor.packaging.utils import NormalizedName, canonicalize_name
from pip._vendor.resolvelib import ResolutionImpossible
from pip._internal.cache import CacheEntry, WheelCache
from pip._internal.exceptions import (
DistributionNotFound,
InstallationError,
MetadataInconsistent,
UnsupportedPythonVersion,
UnsupportedWheel,
)
from pip._internal.index.package_finder import PackageFinder
from pip._internal.metadata import BaseDistribution, get_default_environment
from pip._internal.models.link import Link
from pip._internal.models.wheel import Wheel
from pip._internal.operations.prepare import RequirementPreparer
from pip._internal.req.constructors import (
install_req_drop_extras,
install_req_from_link_and_ireq,
)
from pip._internal.req.req_install import (
InstallRequirement,
check_invalid_constraint_type,
)
from pip._internal.resolution.base import InstallRequirementProvider
from pip._internal.utils.compatibility_tags import get_supported
from pip._internal.utils.hashes import Hashes
from pip._internal.utils.packaging import get_requirement
from pip._internal.utils.virtualenv import running_under_virtualenv
from .base import Candidate, CandidateVersion, Constraint, Requirement
from .candidates import (
AlreadyInstalledCandidate,
BaseCandidate,
EditableCandidate,
ExtrasCandidate,
LinkCandidate,
RequiresPythonCandidate,
as_base_candidate,
)
from .found_candidates import FoundCandidates, IndexCandidateInfo
from .requirements import (
ExplicitRequirement,
RequiresPythonRequirement,
SpecifierRequirement,
SpecifierWithoutExtrasRequirement,
UnsatisfiableRequirement,
)
if TYPE_CHECKING:
from typing import Protocol
class ConflictCause(Protocol):
requirement: RequiresPythonRequirement
parent: Candidate
logger = logging.getLogger(__name__)
C = TypeVar("C")
Cache = Dict[Link, C]
class CollectedRootRequirements(NamedTuple):
requirements: List[Requirement]
constraints: Dict[str, Constraint]
user_requested: Dict[str, int]
class Factory:
def __init__(
self,
finder: PackageFinder,
preparer: RequirementPreparer,
make_install_req: InstallRequirementProvider,
wheel_cache: Optional[WheelCache],
use_user_site: bool,
force_reinstall: bool,
ignore_installed: bool,
ignore_requires_python: bool,
py_version_info: Optional[Tuple[int, ...]] = None,
) -> None:
self._finder = finder
self.preparer = preparer
self._wheel_cache = wheel_cache
self._python_candidate = RequiresPythonCandidate(py_version_info)
self._make_install_req_from_spec = make_install_req
self._use_user_site = use_user_site
self._force_reinstall = force_reinstall
self._ignore_requires_python = ignore_requires_python
self._build_failures: Cache[InstallationError] = {}
self._link_candidate_cache: Cache[LinkCandidate] = {}
self._editable_candidate_cache: Cache[EditableCandidate] = {}
self._installed_candidate_cache: Dict[str, AlreadyInstalledCandidate] = {}
self._extras_candidate_cache: Dict[
Tuple[int, FrozenSet[NormalizedName]], ExtrasCandidate
] = {}
if not ignore_installed:
env = get_default_environment()
self._installed_dists = {
dist.canonical_name: dist
for dist in env.iter_installed_distributions(local_only=False)
}
else:
self._installed_dists = {}
@property
def force_reinstall(self) -> bool:
return self._force_reinstall
def _fail_if_link_is_unsupported_wheel(self, link: Link) -> None:
if not link.is_wheel:
return
wheel = Wheel(link.filename)
if wheel.supported(self._finder.target_python.get_unsorted_tags()):
return
msg = f"{link.filename} is not a supported wheel on this platform."
raise UnsupportedWheel(msg)
def _make_extras_candidate(
self,
base: BaseCandidate,
extras: FrozenSet[str],
*,
comes_from: Optional[InstallRequirement] = None,
) -> ExtrasCandidate:
cache_key = (id(base), frozenset(canonicalize_name(e) for e in extras))
try:
candidate = self._extras_candidate_cache[cache_key]
except KeyError:
candidate = ExtrasCandidate(base, extras, comes_from=comes_from)
self._extras_candidate_cache[cache_key] = candidate
return candidate
def _make_candidate_from_dist(
self,
dist: BaseDistribution,
extras: FrozenSet[str],
template: InstallRequirement,
) -> Candidate:
try:
base = self._installed_candidate_cache[dist.canonical_name]
except KeyError:
base = AlreadyInstalledCandidate(dist, template, factory=self)
self._installed_candidate_cache[dist.canonical_name] = base
if not extras:
return base
return self._make_extras_candidate(base, extras, comes_from=template)
def _make_candidate_from_link(
self,
link: Link,
extras: FrozenSet[str],
template: InstallRequirement,
name: Optional[NormalizedName],
version: Optional[CandidateVersion],
) -> Optional[Candidate]:
base: Optional[BaseCandidate] = self._make_base_candidate_from_link(
link, template, name, version
)
if not extras or base is None:
return base
return self._make_extras_candidate(base, extras, comes_from=template)
def _make_base_candidate_from_link(
self,
link: Link,
template: InstallRequirement,
name: Optional[NormalizedName],
version: Optional[CandidateVersion],
) -> Optional[BaseCandidate]:
if link in self._build_failures:
return None
if template.editable:
if link not in self._editable_candidate_cache:
try:
self._editable_candidate_cache[link] = EditableCandidate(
link,
template,
factory=self,
name=name,
version=version,
)
except MetadataInconsistent as e:
logger.info(
"Discarding [blue underline]%s[/]: [yellow]%s[reset]",
link,
e,
extra={"markup": True},
)
self._build_failures[link] = e
return None
return self._editable_candidate_cache[link]
else:
if link not in self._link_candidate_cache:
try:
self._link_candidate_cache[link] = LinkCandidate(
link,
template,
factory=self,
name=name,
version=version,
)
except MetadataInconsistent as e:
logger.info(
"Discarding [blue underline]%s[/]: [yellow]%s[reset]",
link,
e,
extra={"markup": True},
)
self._build_failures[link] = e
return None
return self._link_candidate_cache[link]
def _iter_found_candidates(
self,
ireqs: Sequence[InstallRequirement],
specifier: SpecifierSet,
hashes: Hashes,
prefers_installed: bool,
incompatible_ids: Set[int],
) -> Iterable[Candidate]:
if not ireqs:
return ()
template = ireqs[0]
assert template.req, "Candidates found on index must be PEP 508"
name = canonicalize_name(template.req.name)
extras: FrozenSet[str] = frozenset()
for ireq in ireqs:
assert ireq.req, "Candidates found on index must be PEP 508"
specifier &= ireq.req.specifier
hashes &= ireq.hashes(trust_internet=False)
extras |= frozenset(ireq.extras)
def _get_installed_candidate() -> Optional[Candidate]:
if self._force_reinstall:
return None
try:
installed_dist = self._installed_dists[name]
except KeyError:
return None
if not specifier.contains(installed_dist.version, prereleases=True):
return None
candidate = self._make_candidate_from_dist(
dist=installed_dist,
extras=extras,
template=template,
)
if id(candidate) in incompatible_ids:
return None
return candidate
def iter_index_candidate_infos() -> Iterator[IndexCandidateInfo]:
result = self._finder.find_best_candidate(
project_name=name,
specifier=specifier,
hashes=hashes,
)
icans = list(result.iter_applicable())
all_yanked = all(ican.link.is_yanked for ican in icans)
def is_pinned(specifier: SpecifierSet) -> bool:
for sp in specifier:
if sp.operator == "===":
return True
if sp.operator != "==":
continue
if sp.version.endswith(".*"):
continue
return True
return False
pinned = is_pinned(specifier)
for ican in reversed(icans):
if not (all_yanked and pinned) and ican.link.is_yanked:
continue
func = functools.partial(
self._make_candidate_from_link,
link=ican.link,
extras=extras,
template=template,
name=name,
version=ican.version,
)
yield ican.version, func
return FoundCandidates(
iter_index_candidate_infos,
_get_installed_candidate(),
prefers_installed,
incompatible_ids,
)
def _iter_explicit_candidates_from_base(
self,
base_requirements: Iterable[Requirement],
extras: FrozenSet[str],
) -> Iterator[Candidate]:
for req in base_requirements:
lookup_cand, _ = req.get_candidate_lookup()
if lookup_cand is None: continue
base_cand = as_base_candidate(lookup_cand)
assert base_cand is not None, "no extras here"
yield self._make_extras_candidate(base_cand, extras)
def _iter_candidates_from_constraints(
self,
identifier: str,
constraint: Constraint,
template: InstallRequirement,
) -> Iterator[Candidate]:
for link in constraint.links:
self._fail_if_link_is_unsupported_wheel(link)
candidate = self._make_base_candidate_from_link(
link,
template=install_req_from_link_and_ireq(link, template),
name=canonicalize_name(identifier),
version=None,
)
if candidate:
yield candidate
def find_candidates(
self,
identifier: str,
requirements: Mapping[str, Iterable[Requirement]],
incompatibilities: Mapping[str, Iterator[Candidate]],
constraint: Constraint,
prefers_installed: bool,
) -> Iterable[Candidate]:
explicit_candidates: Set[Candidate] = set()
ireqs: List[InstallRequirement] = []
for req in requirements[identifier]:
cand, ireq = req.get_candidate_lookup()
if cand is not None:
explicit_candidates.add(cand)
if ireq is not None:
ireqs.append(ireq)
with contextlib.suppress(InvalidRequirement):
parsed_requirement = get_requirement(identifier)
if parsed_requirement.name != identifier:
explicit_candidates.update(
self._iter_explicit_candidates_from_base(
requirements.get(parsed_requirement.name, ()),
frozenset(parsed_requirement.extras),
),
)
for req in requirements.get(parsed_requirement.name, []):
_, ireq = req.get_candidate_lookup()
if ireq is not None:
ireqs.append(ireq)
if ireqs:
try:
explicit_candidates.update(
self._iter_candidates_from_constraints(
identifier,
constraint,
template=ireqs[0],
),
)
except UnsupportedWheel:
return ()
incompat_ids = {id(c) for c in incompatibilities.get(identifier, ())}
if not explicit_candidates:
return self._iter_found_candidates(
ireqs,
constraint.specifier,
constraint.hashes,
prefers_installed,
incompat_ids,
)
return (
c
for c in explicit_candidates
if id(c) not in incompat_ids
and constraint.is_satisfied_by(c)
and all(req.is_satisfied_by(c) for req in requirements[identifier])
)
def _make_requirements_from_install_req(
self, ireq: InstallRequirement, requested_extras: Iterable[str]
) -> Iterator[Requirement]:
if not ireq.match_markers(requested_extras):
logger.info(
"Ignoring %s: markers '%s' don't match your environment",
ireq.name,
ireq.markers,
)
elif not ireq.link:
if ireq.extras and ireq.req is not None and ireq.req.specifier:
yield SpecifierWithoutExtrasRequirement(ireq)
yield SpecifierRequirement(ireq)
else:
self._fail_if_link_is_unsupported_wheel(ireq.link)
cand = self._make_base_candidate_from_link(
ireq.link,
template=install_req_drop_extras(ireq) if ireq.extras else ireq,
name=canonicalize_name(ireq.name) if ireq.name else None,
version=None,
)
if cand is None:
if not ireq.name:
raise self._build_failures[ireq.link]
yield UnsatisfiableRequirement(canonicalize_name(ireq.name))
else:
yield self.make_requirement_from_candidate(cand)
if ireq.extras:
yield self.make_requirement_from_candidate(
self._make_extras_candidate(cand, frozenset(ireq.extras))
)
def collect_root_requirements(
self, root_ireqs: List[InstallRequirement]
) -> CollectedRootRequirements:
collected = CollectedRootRequirements([], {}, {})
for i, ireq in enumerate(root_ireqs):
if ireq.constraint:
problem = check_invalid_constraint_type(ireq)
if problem:
raise InstallationError(problem)
if not ireq.match_markers():
continue
assert ireq.name, "Constraint must be named"
name = canonicalize_name(ireq.name)
if name in collected.constraints:
collected.constraints[name] &= ireq
else:
collected.constraints[name] = Constraint.from_ireq(ireq)
else:
reqs = list(
self._make_requirements_from_install_req(
ireq,
requested_extras=(),
)
)
if not reqs:
continue
template = reqs[0]
if ireq.user_supplied and template.name not in collected.user_requested:
collected.user_requested[template.name] = i
collected.requirements.extend(reqs)
collected.requirements.sort(key=lambda r: r.name != r.project_name)
return collected
def make_requirement_from_candidate(
self, candidate: Candidate
) -> ExplicitRequirement:
return ExplicitRequirement(candidate)
def make_requirements_from_spec(
self,
specifier: str,
comes_from: Optional[InstallRequirement],
requested_extras: Iterable[str] = (),
) -> Iterator[Requirement]:
ireq = self._make_install_req_from_spec(specifier, comes_from)
return self._make_requirements_from_install_req(ireq, requested_extras)
def make_requires_python_requirement(
self,
specifier: SpecifierSet,
) -> Optional[Requirement]:
if self._ignore_requires_python:
return None
if not str(specifier):
return None
return RequiresPythonRequirement(specifier, self._python_candidate)
def get_wheel_cache_entry(
self, link: Link, name: Optional[str]
) -> Optional[CacheEntry]:
if self._wheel_cache is None:
return None
return self._wheel_cache.get_cache_entry(
link=link,
package_name=name,
supported_tags=get_supported(),
)
def get_dist_to_uninstall(self, candidate: Candidate) -> Optional[BaseDistribution]:
dist = self._installed_dists.get(candidate.project_name)
if dist is None: return None
if not self._use_user_site:
return dist
if dist.in_usersite:
return dist
if running_under_virtualenv() and dist.in_site_packages:
message = (
f"Will not install to the user site because it will lack "
f"sys.path precedence to {dist.raw_name} in {dist.location}"
)
raise InstallationError(message)
return None
def _report_requires_python_error(
self, causes: Sequence["ConflictCause"]
) -> UnsupportedPythonVersion:
assert causes, "Requires-Python error reported with no cause"
version = self._python_candidate.version
if len(causes) == 1:
specifier = str(causes[0].requirement.specifier)
message = (
f"Package {causes[0].parent.name!r} requires a different "
f"Python: {version} not in {specifier!r}"
)
return UnsupportedPythonVersion(message)
message = f"Packages require a different Python. {version} not in:"
for cause in causes:
package = cause.parent.format_for_error()
specifier = str(cause.requirement.specifier)
message += f"\n{specifier!r} (required by {package})"
return UnsupportedPythonVersion(message)
def _report_single_requirement_conflict(
self, req: Requirement, parent: Optional[Candidate]
) -> DistributionNotFound:
if parent is None:
req_disp = str(req)
else:
req_disp = f"{req} (from {parent.name})"
cands = self._finder.find_all_candidates(req.project_name)
skipped_by_requires_python = self._finder.requires_python_skipped_reasons()
versions_set: Set[CandidateVersion] = set()
yanked_versions_set: Set[CandidateVersion] = set()
for c in cands:
is_yanked = c.link.is_yanked if c.link else False
if is_yanked:
yanked_versions_set.add(c.version)
else:
versions_set.add(c.version)
versions = [str(v) for v in sorted(versions_set)]
yanked_versions = [str(v) for v in sorted(yanked_versions_set)]
if yanked_versions:
logger.critical(
"Ignored the following yanked versions: %s",
", ".join(yanked_versions) or "none",
)
if skipped_by_requires_python:
logger.critical(
"Ignored the following versions that require a different python "
"version: %s",
"; ".join(skipped_by_requires_python) or "none",
)
logger.critical(
"Could not find a version that satisfies the requirement %s "
"(from versions: %s)",
req_disp,
", ".join(versions) or "none",
)
if str(req) == "requirements.txt":
logger.info(
"HINT: You are attempting to install a package literally "
'named "requirements.txt" (which cannot exist). Consider '
"using the '-r' flag to install the packages listed in "
"requirements.txt"
)
return DistributionNotFound(f"No matching distribution found for {req}")
def get_installation_error(
self,
e: "ResolutionImpossible[Requirement, Candidate]",
constraints: Dict[str, Constraint],
) -> InstallationError:
assert e.causes, "Installation error reported with no cause"
requires_python_causes = [
cause
for cause in e.causes
if isinstance(cause.requirement, RequiresPythonRequirement)
and not cause.requirement.is_satisfied_by(self._python_candidate)
]
if requires_python_causes:
return self._report_requires_python_error(
cast("Sequence[ConflictCause]", requires_python_causes),
)
if len(e.causes) == 1:
req, parent = e.causes[0]
if req.name not in constraints:
return self._report_single_requirement_conflict(req, parent)
def text_join(parts: List[str]) -> str:
if len(parts) == 1:
return parts[0]
return ", ".join(parts[:-1]) + " and " + parts[-1]
def describe_trigger(parent: Candidate) -> str:
ireq = parent.get_install_requirement()
if not ireq or not ireq.comes_from:
return f"{parent.name}=={parent.version}"
if isinstance(ireq.comes_from, InstallRequirement):
return str(ireq.comes_from.name)
return str(ireq.comes_from)
triggers = set()
for req, parent in e.causes:
if parent is None:
trigger = req.format_for_error()
else:
trigger = describe_trigger(parent)
triggers.add(trigger)
if triggers:
info = text_join(sorted(triggers))
else:
info = "the requested packages"
msg = (
f"Cannot install {info} because these package versions "
"have conflicting dependencies."
)
logger.critical(msg)
msg = "\nThe conflict is caused by:"
relevant_constraints = set()
for req, parent in e.causes:
if req.name in constraints:
relevant_constraints.add(req.name)
msg = msg + "\n "
if parent:
msg = msg + f"{parent.name} {parent.version} depends on "
else:
msg = msg + "The user requested "
msg = msg + req.format_for_error()
for key in relevant_constraints:
spec = constraints[key].specifier
msg += f"\n The user requested (constraint) {key}{spec}"
msg = (
msg
+ "\n\n"
+ "To fix this you could try to:\n"
+ "1. loosen the range of package versions you've specified\n"
+ "2. remove package versions to allow pip attempt to solve "
+ "the dependency conflict\n"
)
logger.info(msg)
return DistributionNotFound(
"ResolutionImpossible: for help visit "
"https://pip.pypa.io/en/latest/topics/dependency-resolution/"
"#dealing-with-dependency-conflicts"
)