-
Notifications
You must be signed in to change notification settings - Fork 2.3k
/
pip_installer.py
246 lines (188 loc) · 7.63 KB
/
pip_installer.py
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
import os
import tempfile
from io import open
from subprocess import CalledProcessError
from clikit.api.io import IO
from clikit.io import NullIO
from poetry.repositories.pool import Pool
from poetry.utils._compat import encode
from poetry.utils.env import Env
from poetry.utils.helpers import safe_rmtree
from .base_installer import BaseInstaller
try:
import urllib.parse as urlparse
except ImportError:
import urlparse
class PipInstaller(BaseInstaller):
def __init__(self, env, io, pool): # type: (Env, IO, Pool) -> None
self._env = env
self._io = io
self._pool = pool
def install(self, package, update=False):
if package.source_type == "directory":
self.install_directory(package)
return
if package.source_type == "git":
self.install_git(package)
return
args = ["install", "--no-deps"]
if (
package.source_type not in {"git", "directory", "file", "url"}
and package.source_url
):
repository = self._pool.repository(package.source_reference)
parsed = urlparse.urlparse(package.source_url)
if parsed.scheme == "http":
self._io.error(
" <warning>Installing from unsecure host: {}</warning>".format(
parsed.hostname
)
)
args += ["--trusted-host", parsed.hostname]
if repository.cert:
args += ["--cert", str(repository.cert)]
if repository.client_cert:
args += ["--client-cert", str(repository.client_cert)]
index_url = repository.authenticated_url
args += ["--index-url", index_url]
if self._pool.has_default():
if repository.name != self._pool.repositories[0].name:
args += [
"--extra-index-url",
self._pool.repositories[0].authenticated_url,
]
if update:
args.append("-U")
if package.files and not package.source_url:
# Format as a requirements.txt
# We need to create a requirements.txt file
# for each package in order to check hashes.
# This is far from optimal but we do not have any
# other choice since this is the only way for pip
# to verify hashes.
req = self.create_temporary_requirement(package)
args += ["-r", req]
try:
self.run(*args)
finally:
os.unlink(req)
else:
req = self.requirement(package)
if not isinstance(req, list):
args.append(req)
else:
args += req
self.run(*args)
def update(self, _, target):
self.install(target, update=True)
def remove(self, package):
# If we have a VCS package, remove its source directory
if package.source_type == "git":
src_dir = self._env.path / "src" / package.name
if src_dir.exists():
safe_rmtree(str(src_dir))
try:
self.run("uninstall", package.name, "-y")
except CalledProcessError as e:
if "not installed" in str(e):
return
raise
def run(self, *args, **kwargs): # type: (...) -> str
return self._env.run_pip(*args, **kwargs)
def requirement(self, package, formatted=False):
if formatted and not package.source_type:
req = "{}=={}".format(package.name, package.version)
for f in package.files:
hash_type = "sha256"
h = f["hash"]
if ":" in h:
hash_type, h = h.split(":")
req += " --hash {}:{}".format(hash_type, h)
req += "\n"
return req
if package.source_type in ["file", "directory"]:
if package.root_dir:
req = os.path.join(package.root_dir, package.source_url)
else:
req = os.path.realpath(package.source_url)
if package.develop and package.source_type == "directory":
req = ["-e", req]
return req
if package.source_type == "git":
req = "git+{}@{}#egg={}".format(
package.source_url, package.source_reference, package.name
)
if package.develop:
req = ["-e", req]
return req
if package.source_type == "url":
return "{}#egg={}".format(package.source_url, package.name)
return "{}=={}".format(package.name, package.version)
def create_temporary_requirement(self, package):
fd, name = tempfile.mkstemp(
"reqs.txt", "{}-{}".format(package.name, package.version)
)
try:
os.write(fd, encode(self.requirement(package, formatted=True)))
finally:
os.close(fd)
return name
def install_directory(self, package):
from poetry.masonry.builder import SdistBuilder
from poetry.factory import Factory
from poetry.utils._compat import decode
from poetry.utils.env import NullEnv
from poetry.utils.toml_file import TomlFile
if package.root_dir:
req = os.path.join(package.root_dir, package.source_url)
else:
req = os.path.realpath(package.source_url)
args = ["install", "--no-deps", "-U"]
pyproject = TomlFile(os.path.join(req, "pyproject.toml"))
has_poetry = False
has_build_system = False
if pyproject.exists():
pyproject_content = pyproject.read()
has_poetry = (
"tool" in pyproject_content and "poetry" in pyproject_content["tool"]
)
# Even if there is a build system specified
# pip as of right now does not support it fully
# TODO: Check for pip version when proper PEP-517 support lands
# has_build_system = ("build-system" in pyproject_content)
setup = os.path.join(req, "setup.py")
has_setup = os.path.exists(setup)
if not has_setup and has_poetry and (package.develop or not has_build_system):
# We actually need to rely on creating a temporary setup.py
# file since pip, as of this comment, does not support
# build-system for editable packages
# We also need it for non-PEP-517 packages
builder = SdistBuilder(
Factory().create_poetry(pyproject.parent), NullEnv(), NullIO()
)
with open(setup, "w", encoding="utf-8") as f:
f.write(decode(builder.build_setup()))
if package.develop:
args.append("-e")
args.append(req)
try:
return self.run(*args)
finally:
if not has_setup and os.path.exists(setup):
os.remove(setup)
def install_git(self, package):
from poetry.packages import Package
from poetry.vcs import Git
src_dir = self._env.path / "src" / package.name
if src_dir.exists():
safe_rmtree(str(src_dir))
src_dir.parent.mkdir(exist_ok=True)
git = Git()
git.clone(package.source_url, src_dir)
git.checkout(package.source_reference, src_dir)
# Now we just need to install from the source directory
pkg = Package(package.name, package.version)
pkg.source_type = "directory"
pkg.source_url = str(src_dir)
pkg.develop = package.develop
self.install_directory(pkg)