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
|
from pip._vendor.packaging.utils import canonicalize_name
from pip._internal.exceptions import (
InstallationError,
UnsupportedPythonVersion,
)
from pip._internal.utils.misc import get_installed_distributions
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
from .candidates import (
AlreadyInstalledCandidate,
EditableCandidate,
ExtrasCandidate,
LinkCandidate,
RequiresPythonCandidate,
)
from .requirements import (
ExplicitRequirement,
RequiresPythonRequirement,
SpecifierRequirement,
)
if MYPY_CHECK_RUNNING:
from typing import Dict, Iterator, Optional, Set, Tuple, TypeVar
from pip._vendor.packaging.specifiers import SpecifierSet
from pip._vendor.packaging.version import _BaseVersion
from pip._vendor.pkg_resources import Distribution
from pip._vendor.resolvelib import ResolutionImpossible
from pip._internal.index.package_finder import PackageFinder
from pip._internal.models.link import Link
from pip._internal.operations.prepare import RequirementPreparer
from pip._internal.req.req_install import InstallRequirement
from pip._internal.resolution.base import InstallRequirementProvider
from .base import Candidate, Requirement
from .candidates import BaseCandidate
C = TypeVar("C")
Cache = Dict[Link, C]
class Factory(object):
def __init__(
self,
finder, # type: PackageFinder
preparer, # type: RequirementPreparer
make_install_req, # type: InstallRequirementProvider
force_reinstall, # type: bool
ignore_installed, # type: bool
ignore_requires_python, # type: bool
py_version_info=None, # type: Optional[Tuple[int, ...]]
):
# type: (...) -> None
self.finder = finder
self.preparer = preparer
self._python_candidate = RequiresPythonCandidate(py_version_info)
self._make_install_req_from_spec = make_install_req
self._force_reinstall = force_reinstall
self._ignore_requires_python = ignore_requires_python
self._link_candidate_cache = {} # type: Cache[LinkCandidate]
self._editable_candidate_cache = {} # type: Cache[EditableCandidate]
if not ignore_installed:
self._installed_dists = {
canonicalize_name(dist.project_name): dist
for dist in get_installed_distributions()
}
else:
self._installed_dists = {}
def _make_candidate_from_dist(
self,
dist, # type: Distribution
extras, # type: Set[str]
parent, # type: InstallRequirement
):
# type: (...) -> Candidate
base = AlreadyInstalledCandidate(dist, parent, factory=self)
if extras:
return ExtrasCandidate(base, extras)
return base
def _make_candidate_from_link(
self,
link, # type: Link
extras, # type: Set[str]
parent, # type: InstallRequirement
name=None, # type: Optional[str]
version=None, # type: Optional[_BaseVersion]
):
# type: (...) -> Candidate
# TODO: Check already installed candidate, and use it if the link and
# editable flag match.
if parent.editable:
if link not in self._editable_candidate_cache:
self._editable_candidate_cache[link] = EditableCandidate(
link, parent, factory=self, name=name, version=version,
)
base = self._editable_candidate_cache[link] # type: BaseCandidate
else:
if link not in self._link_candidate_cache:
self._link_candidate_cache[link] = LinkCandidate(
link, parent, factory=self, name=name, version=version,
)
base = self._link_candidate_cache[link]
if extras:
return ExtrasCandidate(base, extras)
return base
def iter_found_candidates(self, ireq, extras):
# type: (InstallRequirement, Set[str]) -> Iterator[Candidate]
name = canonicalize_name(ireq.req.name)
if not self._force_reinstall:
installed_dist = self._installed_dists.get(name)
else:
installed_dist = None
found = self.finder.find_best_candidate(
project_name=ireq.req.name,
specifier=ireq.req.specifier,
hashes=ireq.hashes(trust_internet=False),
)
for ican in found.iter_applicable():
if (installed_dist is not None and
installed_dist.parsed_version == ican.version):
continue
yield self._make_candidate_from_link(
link=ican.link,
extras=extras,
parent=ireq,
name=name,
version=ican.version,
)
# Return installed distribution if it matches the specifier. This is
# done last so the resolver will prefer it over downloading links.
if (installed_dist is not None and
installed_dist.parsed_version in ireq.req.specifier):
yield self._make_candidate_from_dist(
dist=installed_dist,
extras=extras,
parent=ireq,
)
def make_requirement_from_install_req(self, ireq):
# type: (InstallRequirement) -> Requirement
if ireq.link:
# TODO: Get name and version from ireq, if possible?
# Specifically, this might be needed in "name @ URL"
# syntax - need to check where that syntax is handled.
cand = self._make_candidate_from_link(
ireq.link, extras=set(), parent=ireq,
)
return ExplicitRequirement(cand)
return SpecifierRequirement(ireq, factory=self)
def make_requirement_from_spec(self, specifier, comes_from):
# type: (str, InstallRequirement) -> Requirement
ireq = self._make_install_req_from_spec(specifier, comes_from)
return self.make_requirement_from_install_req(ireq)
def make_requires_python_requirement(self, specifier):
# type: (Optional[SpecifierSet]) -> Optional[Requirement]
if self._ignore_requires_python or specifier is None:
return None
return RequiresPythonRequirement(specifier, self._python_candidate)
def should_reinstall(self, candidate):
# type: (Candidate) -> bool
# TODO: Are there more cases this needs to return True? Editable?
return candidate.name in self._installed_dists
def _report_requires_python_error(
self,
requirement, # type: RequiresPythonRequirement
parent, # type: Candidate
):
# type: (...) -> UnsupportedPythonVersion
template = (
"Package {package!r} requires a different Python: "
"{version} not in {specifier!r}"
)
message = template.format(
package=parent.name,
version=self._python_candidate.version,
specifier=str(requirement.specifier),
)
return UnsupportedPythonVersion(message)
def get_installation_error(self, e):
# type: (ResolutionImpossible) -> Optional[InstallationError]
for cause in e.causes:
if isinstance(cause.requirement, RequiresPythonRequirement):
return self._report_requires_python_error(
cause.requirement,
cause.parent,
)
return None
|
