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
|
from __future__ import unicode_literals
import os
import re
from itertools import chain
from pip._vendor import six
from .click import unstyle
from .logging import log
from .utils import (
UNSAFE_PACKAGES,
comment,
dedup,
format_requirement,
get_compile_command,
key_from_ireq,
)
MESSAGE_UNHASHED_PACKAGE = comment(
"# WARNING: pip install will require the following package to be hashed."
"\n# Consider using a hashable URL like "
"https://github.com/jazzband/pip-tools/archive/SOMECOMMIT.zip"
)
MESSAGE_UNSAFE_PACKAGES_UNPINNED = comment(
"# WARNING: The following packages were not pinned, but pip requires them to be"
"\n# pinned when the requirements file includes hashes. "
"Consider using the --allow-unsafe flag."
)
MESSAGE_UNSAFE_PACKAGES = comment(
"# The following packages are considered to be unsafe in a requirements file:"
)
MESSAGE_UNINSTALLABLE = (
"The generated requirements file may be rejected by pip install. "
"See # WARNING lines for details."
)
strip_comes_from_line_re = re.compile(r" \(line \d+\)$")
def _comes_from_as_string(ireq):
if isinstance(ireq.comes_from, six.string_types):
return strip_comes_from_line_re.sub("", ireq.comes_from)
return key_from_ireq(ireq.comes_from)
class OutputWriter(object):
def __init__(
self,
src_files,
dst_file,
click_ctx,
dry_run,
emit_header,
emit_index_url,
emit_trusted_host,
annotate,
generate_hashes,
default_index_url,
index_urls,
trusted_hosts,
format_control,
allow_unsafe,
find_links,
emit_find_links,
):
self.src_files = src_files
self.dst_file = dst_file
self.click_ctx = click_ctx
self.dry_run = dry_run
self.emit_header = emit_header
self.emit_index_url = emit_index_url
self.emit_trusted_host = emit_trusted_host
self.annotate = annotate
self.generate_hashes = generate_hashes
self.default_index_url = default_index_url
self.index_urls = index_urls
self.trusted_hosts = trusted_hosts
self.format_control = format_control
self.allow_unsafe = allow_unsafe
self.find_links = find_links
self.emit_find_links = emit_find_links
def _sort_key(self, ireq):
return (not ireq.editable, str(ireq.req).lower())
def write_header(self):
if self.emit_header:
yield comment("#")
yield comment("# This file is autogenerated by pip-compile")
yield comment("# To update, run:")
yield comment("#")
compile_command = os.environ.get(
"CUSTOM_COMPILE_COMMAND"
) or get_compile_command(self.click_ctx)
yield comment("# {}".format(compile_command))
yield comment("#")
def write_index_options(self):
if self.emit_index_url:
for index, index_url in enumerate(dedup(self.index_urls)):
if index_url.rstrip("/") == self.default_index_url:
continue
flag = "--index-url" if index == 0 else "--extra-index-url"
yield "{} {}".format(flag, index_url)
def write_trusted_hosts(self):
if self.emit_trusted_host:
for trusted_host in dedup(self.trusted_hosts):
yield "--trusted-host {}".format(trusted_host)
def write_format_controls(self):
for nb in dedup(sorted(self.format_control.no_binary)):
yield "--no-binary {}".format(nb)
for ob in dedup(sorted(self.format_control.only_binary)):
yield "--only-binary {}".format(ob)
def write_find_links(self):
if self.emit_find_links:
for find_link in dedup(self.find_links):
yield "--find-links {}".format(find_link)
def write_flags(self):
emitted = False
for line in chain(
self.write_index_options(),
self.write_find_links(),
self.write_trusted_hosts(),
self.write_format_controls(),
):
emitted = True
yield line
if emitted:
yield ""
def _iter_lines(self, results, unsafe_requirements=None, markers=None, hashes=None):
# default values
unsafe_requirements = unsafe_requirements or []
markers = markers or {}
hashes = hashes or {}
# Check for unhashed or unpinned packages if at least one package does have
# hashes, which will trigger pip install's --require-hashes mode.
warn_uninstallable = False
has_hashes = hashes and any(hash for hash in hashes.values())
yielded = False
for line in self.write_header():
yield line
yielded = True
for line in self.write_flags():
yield line
yielded = True
unsafe_requirements = (
{r for r in results if r.name in UNSAFE_PACKAGES}
if not unsafe_requirements
else unsafe_requirements
)
packages = {r for r in results if r.name not in UNSAFE_PACKAGES}
if packages:
packages = sorted(packages, key=self._sort_key)
for ireq in packages:
if has_hashes and not hashes.get(ireq):
yield MESSAGE_UNHASHED_PACKAGE
warn_uninstallable = True
line = self._format_requirement(
ireq, markers.get(key_from_ireq(ireq)), hashes=hashes
)
yield line
yielded = True
if unsafe_requirements:
unsafe_requirements = sorted(unsafe_requirements, key=self._sort_key)
yield ""
yielded = True
if has_hashes and not self.allow_unsafe:
yield MESSAGE_UNSAFE_PACKAGES_UNPINNED
warn_uninstallable = True
else:
yield MESSAGE_UNSAFE_PACKAGES
for ireq in unsafe_requirements:
ireq_key = key_from_ireq(ireq)
if not self.allow_unsafe:
yield comment("# {}".format(ireq_key))
else:
line = self._format_requirement(
ireq, marker=markers.get(ireq_key), hashes=hashes
)
yield line
# Yield even when there's no real content, so that blank files are written
if not yielded:
yield ""
if warn_uninstallable:
log.warning(MESSAGE_UNINSTALLABLE)
def write(self, results, unsafe_requirements, markers, hashes):
for line in self._iter_lines(results, unsafe_requirements, markers, hashes):
log.info(line)
if not self.dry_run:
self.dst_file.write(unstyle(line).encode("utf-8"))
self.dst_file.write(os.linesep.encode("utf-8"))
def _format_requirement(self, ireq, marker=None, hashes=None):
ireq_hashes = (hashes if hashes is not None else {}).get(ireq)
line = format_requirement(ireq, marker=marker, hashes=ireq_hashes)
if not self.annotate:
return line
# Annotate what packages or reqs-ins this package is required by
required_by = set()
if hasattr(ireq, "_source_ireqs"):
required_by |= {
_comes_from_as_string(src_ireq)
for src_ireq in ireq._source_ireqs
if src_ireq.comes_from
}
elif ireq.comes_from:
required_by.add(_comes_from_as_string(ireq))
if required_by:
required_by = sorted(required_by)
if len(required_by) == 1:
source = required_by[0]
annotation = " # via " + source
else:
annotation_lines = [" # via"]
for source in required_by:
annotation_lines.append(" # " + source)
annotation = "\n".join(annotation_lines)
line = "{}\n{}".format(line, comment(annotation))
return line
|
