Mercurial > repos > guerler > hhblits

comparison lib/python3.8/site-packages/pip/_internal/req/req_set.py @ 0:9e54283cc701 draft

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
"planemo upload commit d12c32a45bcd441307e632fca6d9af7d60289d44"
author guerler
date Mon, 27 Jul 2020 03:47:31 -0400
parents
children
comparison
equal deleted inserted replaced
-1:000000000000 0:9e54283cc701
1 # The following comment should be removed at some point in the future.
2 # mypy: strict-optional=False
3
4 from __future__ import absolute_import
5
6 import logging
7 from collections import OrderedDict
8
9 from pip._vendor.packaging.utils import canonicalize_name
10
11 from pip._internal import pep425tags
12 from pip._internal.exceptions import InstallationError
13 from pip._internal.models.wheel import Wheel
14 from pip._internal.utils.logging import indent_log
15 from pip._internal.utils.typing import MYPY_CHECK_RUNNING
16
17 if MYPY_CHECK_RUNNING:
18 from typing import Dict, Iterable, List, Optional, Tuple
19 from pip._internal.req.req_install import InstallRequirement
20
21
22 logger = logging.getLogger(__name__)
23
24
25 class RequirementSet(object):
26
27 def __init__(self, check_supported_wheels=True):
28 # type: (bool) -> None
29 """Create a RequirementSet.
30 """
31
32 self.requirements = OrderedDict() # type: Dict[str, InstallRequirement] # noqa: E501
33 self.check_supported_wheels = check_supported_wheels
34
35 self.unnamed_requirements = [] # type: List[InstallRequirement]
36 self.successfully_downloaded = [] # type: List[InstallRequirement]
37 self.reqs_to_cleanup = [] # type: List[InstallRequirement]
38
39 def __str__(self):
40 # type: () -> str
41 requirements = sorted(
42 (req for req in self.requirements.values() if not req.comes_from),
43 key=lambda req: canonicalize_name(req.name),
44 )
45 return ' '.join(str(req.req) for req in requirements)
46
47 def __repr__(self):
48 # type: () -> str
49 requirements = sorted(
50 self.requirements.values(),
51 key=lambda req: canonicalize_name(req.name),
52 )
53
54 format_string = '<{classname} object; {count} requirement(s): {reqs}>'
55 return format_string.format(
56 classname=self.__class__.__name__,
57 count=len(requirements),
58 reqs=', '.join(str(req.req) for req in requirements),
59 )
60
61 def add_unnamed_requirement(self, install_req):
62 # type: (InstallRequirement) -> None
63 assert not install_req.name
64 self.unnamed_requirements.append(install_req)
65
66 def add_named_requirement(self, install_req):
67 # type: (InstallRequirement) -> None
68 assert install_req.name
69
70 project_name = canonicalize_name(install_req.name)
71 self.requirements[project_name] = install_req
72
73 def add_requirement(
74 self,
75 install_req, # type: InstallRequirement
76 parent_req_name=None, # type: Optional[str]
77 extras_requested=None # type: Optional[Iterable[str]]
78 ):
79 # type: (...) -> Tuple[List[InstallRequirement], Optional[InstallRequirement]] # noqa: E501
80 """Add install_req as a requirement to install.
81
82 :param parent_req_name: The name of the requirement that needed this
83 added. The name is used because when multiple unnamed requirements
84 resolve to the same name, we could otherwise end up with dependency
85 links that point outside the Requirements set. parent_req must
86 already be added. Note that None implies that this is a user
87 supplied requirement, vs an inferred one.
88 :param extras_requested: an iterable of extras used to evaluate the
89 environment markers.
90 :return: Additional requirements to scan. That is either [] if
91 the requirement is not applicable, or [install_req] if the
92 requirement is applicable and has just been added.
93 """
94 # If the markers do not match, ignore this requirement.
95 if not install_req.match_markers(extras_requested):
96 logger.info(
97 "Ignoring %s: markers '%s' don't match your environment",
98 install_req.name, install_req.markers,
99 )
100 return [], None
101
102 # If the wheel is not supported, raise an error.
103 # Should check this after filtering out based on environment markers to
104 # allow specifying different wheels based on the environment/OS, in a
105 # single requirements file.
106 if install_req.link and install_req.link.is_wheel:
107 wheel = Wheel(install_req.link.filename)
108 tags = pep425tags.get_supported()
109 if (self.check_supported_wheels and not wheel.supported(tags)):
110 raise InstallationError(
111 "%s is not a supported wheel on this platform." %
112 wheel.filename
113 )
114
115 # This next bit is really a sanity check.
116 assert install_req.is_direct == (parent_req_name is None), (
117 "a direct req shouldn't have a parent and also, "
118 "a non direct req should have a parent"
119 )
120
121 # Unnamed requirements are scanned again and the requirement won't be
122 # added as a dependency until after scanning.
123 if not install_req.name:
124 self.add_unnamed_requirement(install_req)
125 return [install_req], None
126
127 try:
128 existing_req = self.get_requirement(install_req.name)
129 except KeyError:
130 existing_req = None
131
132 has_conflicting_requirement = (
133 parent_req_name is None and
134 existing_req and
135 not existing_req.constraint and
136 existing_req.extras == install_req.extras and
137 existing_req.req.specifier != install_req.req.specifier
138 )
139 if has_conflicting_requirement:
140 raise InstallationError(
141 "Double requirement given: %s (already in %s, name=%r)"
142 % (install_req, existing_req, install_req.name)
143 )
144
145 # When no existing requirement exists, add the requirement as a
146 # dependency and it will be scanned again after.
147 if not existing_req:
148 self.add_named_requirement(install_req)
149 # We'd want to rescan this requirement later
150 return [install_req], install_req
151
152 # Assume there's no need to scan, and that we've already
153 # encountered this for scanning.
154 if install_req.constraint or not existing_req.constraint:
155 return [], existing_req
156
157 does_not_satisfy_constraint = (
158 install_req.link and
159 not (
160 existing_req.link and
161 install_req.link.path == existing_req.link.path
162 )
163 )
164 if does_not_satisfy_constraint:
165 self.reqs_to_cleanup.append(install_req)
166 raise InstallationError(
167 "Could not satisfy constraints for '%s': "
168 "installation from path or url cannot be "
169 "constrained to a version" % install_req.name,
170 )
171 # If we're now installing a constraint, mark the existing
172 # object for real installation.
173 existing_req.constraint = False
174 existing_req.extras = tuple(sorted(
175 set(existing_req.extras) | set(install_req.extras)
176 ))
177 logger.debug(
178 "Setting %s extras to: %s",
179 existing_req, existing_req.extras,
180 )
181 # Return the existing requirement for addition to the parent and
182 # scanning again.
183 return [existing_req], existing_req
184
185 def has_requirement(self, name):
186 # type: (str) -> bool
187 project_name = canonicalize_name(name)
188
189 return (
190 project_name in self.requirements and
191 not self.requirements[project_name].constraint
192 )
193
194 def get_requirement(self, name):
195 # type: (str) -> InstallRequirement
196 project_name = canonicalize_name(name)
197
198 if project_name in self.requirements:
199 return self.requirements[project_name]
200
201 raise KeyError("No project with the name %r" % name)
202
203 def cleanup_files(self):
204 # type: () -> None
205 """Clean up files, remove builds."""
206 logger.debug('Cleaning up...')
207 with indent_log():
208 for req in self.reqs_to_cleanup:
209 req.remove_temporary_source()