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
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
|
"""Adapted from the pip code base"""
from __future__ import annotations
import os
import re
import shlex
import sys
import urllib.parse
from argparse import ArgumentParser, Namespace
from pathlib import Path
from typing import IO, Any, Iterator, List, Tuple, cast
from urllib.request import urlopen
import chardet
from packaging.requirements import InvalidRequirement, Requirement
from .args import build_parser
from .util import VCS, get_url_scheme, is_url, url_to_path
# Matches environment variable-style values in '${MY_VARIABLE_1}' with the variable name consisting of only uppercase
# letters, digits or the '_' (underscore). This follows the POSIX standard defined in IEEE Std 1003.1, 2013 Edition.
_ENV_VAR_RE = re.compile(r"(?P<var>\${(?P<name>[A-Z0-9_]+)})")
_SCHEME_RE = re.compile(r"^(http|https|file):", re.I)
_COMMENT_RE = re.compile(r"(^|\s+)#.*$")
# https://www.python.org/dev/peps/pep-0508/#extras
_EXTRA_PATH = re.compile(r"(.*)\[([-._,\sa-zA-Z0-9]*)]")
_EXTRA_ELEMENT = re.compile(r"[a-zA-Z0-9]*[-._a-zA-Z0-9]")
ReqFileLines = Iterator[Tuple[int, str]]
DEFAULT_INDEX_URL = "https://pypi.org/simple"
class ParsedRequirement:
def __init__(self, req: str, options: dict[str, Any], from_file: str, lineno: int) -> None:
req = req.encode("utf-8").decode("utf-8")
try:
self._requirement: Requirement | Path | str = Requirement(req)
except InvalidRequirement:
if is_url(req) or any(req.startswith(f"{v}+") and is_url(req[len(v) + 1 :]) for v in VCS):
self._requirement = req
else:
root = Path(from_file).parent
extras: list[str] = []
match = _EXTRA_PATH.fullmatch(Path(req).name)
if match:
for extra in match.group(2).split(","):
extra = extra.strip()
if not extra:
continue
if not _EXTRA_ELEMENT.fullmatch(extra):
extras = []
path = root / req
break
extras.append(extra)
else:
path = root / Path(req).parent / match.group(1)
else:
path = root / req
extra_part = f"[{','.join(sorted(extras))}]" if extras else ""
try:
rel_path = str(path.resolve().relative_to(root))
# prefix paths in cwd to not convert them to requirement
if rel_path != "." and os.sep not in rel_path:
rel_path = f".{os.sep}{rel_path}"
except ValueError:
rel_path = str(path.resolve())
self._requirement = f"{rel_path}{extra_part}"
self._options = options
self._from_file = from_file
self._lineno = lineno
@property
def requirement(self) -> Requirement | Path | str:
return self._requirement
@property
def from_file(self) -> str:
return self._from_file
@property
def lineno(self) -> int:
return self._lineno
@property
def options(self) -> dict[str, Any]:
return self._options
def __repr__(self) -> str:
base = f"{self.__class__.__name__}(requirement={self._requirement}, "
if self._options:
base += f"options={self._options!r}, "
return f"{base.rstrip(', ')})"
def __str__(self) -> str:
result = []
if self.options.get("is_constraint"):
result.append("-c")
if self.options.get("is_editable"):
result.append("-e")
result.append(str(self.requirement))
for hash_value in self.options.get("hash", []):
result.extend(("--hash", hash_value))
return " ".join(result)
def as_args(self) -> Iterator[str]:
if self.options.get("is_editable"):
yield "-e"
yield str(self._requirement)
class ParsedLine:
def __init__(self, filename: str, lineno: int, args: str, opts: Namespace, constraint: bool) -> None:
self.filename = filename
self.lineno = lineno
self.opts = opts
self.constraint = constraint
if args:
self.is_requirement = True
self.is_editable = False
self.requirement = args
elif opts.editables:
self.is_requirement = True
self.is_editable = True
# We don't support multiple -e on one line
self.requirement = opts.editables[0]
else:
self.is_requirement = False
class RequirementsFile:
def __init__(self, path: Path, constraint: bool) -> None:
self._path = path
self._is_constraint: bool = constraint
self._opt = Namespace()
self._requirements: list[ParsedRequirement] | None = None
self._as_root_args: list[str] | None = None
self._parser_private: ArgumentParser | None = None
@property
def _req_parser(self) -> RequirementsFile:
return self
def __str__(self) -> str:
return f"{'-c' if self.is_constraint else '-r'} {self.path}"
@property
def path(self) -> Path:
return self._path
@property
def is_constraint(self) -> bool:
return self._is_constraint
@property
def options(self) -> Namespace:
self._ensure_requirements_parsed()
return self._opt
@property
def requirements(self) -> list[ParsedRequirement]:
self._ensure_requirements_parsed()
return cast(List[ParsedRequirement], self._requirements)
@property
def _parser(self) -> ArgumentParser:
if self._parser_private is None:
self._parser_private = build_parser()
self._extend_parser(self._parser_private)
return self._parser_private
def _extend_parser(self, parser: ArgumentParser) -> None: # noqa: U100
...
def _ensure_requirements_parsed(self) -> None:
if self._requirements is None:
self._requirements = self._parse_requirements(opt=self._opt, recurse=True)
def _parse_requirements(self, opt: Namespace, recurse: bool) -> list[ParsedRequirement]:
result, found = [], set()
for parsed_line in self._parse_and_recurse(str(self._path), self.is_constraint, recurse):
if parsed_line.is_requirement:
parsed_req = self._handle_requirement_line(parsed_line)
key = str(parsed_req)
if key not in found:
found.add(key)
result.append(parsed_req)
else:
self._merge_option_line(opt, parsed_line.opts, parsed_line.filename)
result.sort(key=self._key_func)
return result
def _key_func(self, line: ParsedRequirement) -> tuple[int, tuple[int, str, str]]:
of_type = {Requirement: 0, Path: 1, str: 2}[type(line.requirement)]
between = of_type, str(line.requirement).lower(), str(line.options)
if "is_constraint" in line.options:
return 2, between
if "is_editable" in line.options:
return 1, between
return 0, between
def _parse_and_recurse(self, filename: str, constraint: bool, recurse: bool) -> Iterator[ParsedLine]:
for line in self._parse_file(filename, constraint):
if not line.is_requirement and (line.opts.requirements or line.opts.constraints):
if line.opts.requirements: # parse a nested requirements file
nested_constraint, req_path = False, line.opts.requirements[0]
else:
nested_constraint, req_path = True, line.opts.constraints[0]
if _SCHEME_RE.search(filename): # original file is over http
req_path = urllib.parse.urljoin(filename, req_path) # do a url join so relative paths work
elif not _SCHEME_RE.search(req_path): # original file and nested file are paths
# do a join so relative paths work
req_path = os.path.join(os.path.dirname(filename), req_path)
if recurse:
yield from self._req_parser._parse_and_recurse(req_path, nested_constraint, recurse)
else:
line.filename = req_path
yield line
else:
yield line
def _parse_file(self, url: str, constraint: bool) -> Iterator[ParsedLine]:
content = self._get_file_content(url)
for line_number, line in self._pre_process(content):
args_str, opts = self._parse_line(line)
yield ParsedLine(url, line_number, args_str, opts, constraint)
def _get_file_content(self, url: str) -> str:
"""
Gets the content of a file; it may be a filename, file: URL, or http: URL. Returns (location, content).
Content is unicode. Respects # -*- coding: declarations on the retrieved files.
:param url: File path or url.
"""
scheme = get_url_scheme(url)
if scheme in ["http", "https"]:
with urlopen(url) as response:
text = self._read_decode(response)
return text
elif scheme == "file":
url = url_to_path(url)
try:
with open(url, "rb") as file_handler:
text = self._read_decode(file_handler)
except OSError as exc:
raise ValueError(f"Could not open requirements file {url}: {exc}") from exc
return text
@staticmethod
def _read_decode(file_handler: IO[bytes]) -> str:
raw = file_handler.read()
if not raw:
return ""
codec = chardet.detect(raw)["encoding"]
text = raw.decode(codec)
return text
def _pre_process(self, content: str) -> ReqFileLines:
"""Split, filter, and join lines, and return a line iterator
:param content: the content of the requirements file
"""
lines_enum: ReqFileLines = enumerate(content.splitlines(), start=1)
lines_enum = self._join_lines(lines_enum)
lines_enum = self._ignore_comments(lines_enum)
lines_enum = self._expand_env_variables(lines_enum)
return lines_enum
def _parse_line(self, line: str) -> tuple[str, Namespace]:
args_str, options_str = self._break_args_options(line)
args = shlex.split(options_str, posix=sys.platform != "win32")
opts = self._parser.parse_args(args)
return args_str, opts
@staticmethod
def _handle_requirement_line(line: ParsedLine) -> ParsedRequirement:
# For editable requirements, we don't support per-requirement options, so just return the parsed requirement.
# get the options that apply to requirements
req_options: dict[str, Any] = {}
if line.is_editable:
req_options["is_editable"] = line.is_editable
if line.constraint:
req_options["is_constraint"] = line.constraint
hash_values = getattr(line.opts, "hash", [])
if hash_values:
req_options["hash"] = hash_values
return ParsedRequirement(line.requirement, req_options, line.filename, line.lineno)
def _merge_option_line(self, base_opt: Namespace, opt: Namespace, filename: str) -> None: # noqa: C901
# percolate options upward
if opt.requirements:
if not hasattr(base_opt, "requirements"):
base_opt.requirements = []
if opt.requirements[0] not in base_opt.requirements:
base_opt.requirements.append(opt.requirements[0])
if opt.constraints:
if not hasattr(base_opt, "constraints"):
base_opt.constraints = []
if opt.constraints[0] not in base_opt.constraints:
base_opt.constraints.append(opt.constraints[0])
if opt.require_hashes:
base_opt.require_hashes = True
if opt.features_enabled:
if not hasattr(base_opt, "features_enabled"):
base_opt.features_enabled = []
for feature in opt.features_enabled:
if feature not in base_opt.features_enabled:
base_opt.features_enabled.append(feature)
base_opt.features_enabled.sort()
if opt.index_url:
if getattr(base_opt, "index_url", []):
base_opt.index_url[0] = opt.index_url
else:
base_opt.index_url = [opt.index_url]
if opt.no_index is True:
base_opt.index_url = []
if opt.extra_index_url:
if not getattr(base_opt, "index_url", []):
base_opt.index_url = [DEFAULT_INDEX_URL]
for url in opt.extra_index_url:
if url not in base_opt.index_url:
base_opt.index_url.extend(opt.extra_index_url)
if opt.find_links:
# FIXME: it would be nice to keep track of the source of the find_links: support a find-links local path
# relative to a requirements file.
if not hasattr(base_opt, "index_url"): # pragma: no branch
base_opt.find_links = []
value = opt.find_links[0]
req_dir = os.path.dirname(os.path.abspath(filename))
relative_to_reqs_file = os.path.join(req_dir, value)
if os.path.exists(relative_to_reqs_file):
value = relative_to_reqs_file # pragma: no cover
if value not in base_opt.find_links: # pragma: no branch
base_opt.find_links.append(value)
if opt.pre:
base_opt.pre = True
if opt.prefer_binary:
base_opt.prefer_binary = True
for host in opt.trusted_host or []:
if not hasattr(base_opt, "trusted_hosts"):
base_opt.trusted_hosts = []
if host not in base_opt.trusted_hosts:
base_opt.trusted_hosts.append(host)
if opt.no_binary:
base_opt.no_binary = opt.no_binary
if opt.only_binary:
base_opt.only_binary = opt.only_binary
@staticmethod
def _break_args_options(line: str) -> tuple[str, str]:
"""
Break up the line into an args and options string. We only want to shlex (and then optparse) the options, not
the args. args can contain markers which are corrupted by shlex.
"""
tokens = line.split(" ")
args = []
options = tokens[:]
for token in tokens:
if token.startswith("-"): # both `-` and `--` accepted
break
else:
args.append(token)
options.pop(0)
return " ".join(args).strip(), " ".join(options)
@staticmethod
def _join_lines(lines_enum: ReqFileLines) -> ReqFileLines:
"""
Joins a line ending in '\' with the previous line (except when following comments). The joined line takes on the
index of the first line.
"""
primary_line_number = None
new_line: list[str] = []
for line_number, line in lines_enum:
if not line.endswith("\\") or _COMMENT_RE.match(line):
if _COMMENT_RE.match(line):
line = f" {line}" # this ensures comments are always matched later
if new_line:
new_line.append(line)
assert primary_line_number is not None
yield primary_line_number, "".join(new_line)
new_line = []
else:
yield line_number, line
else:
if not new_line: # pragma: no branch
primary_line_number = line_number
new_line.append(line.strip("\\"))
# last line contains \
if new_line:
assert primary_line_number is not None
yield primary_line_number, "".join(new_line)
@staticmethod
def _ignore_comments(lines_enum: ReqFileLines) -> ReqFileLines:
"""Strips comments and filter empty lines."""
for line_number, line in lines_enum:
line = _COMMENT_RE.sub("", line)
line = line.strip()
if line:
yield line_number, line
@staticmethod
def _expand_env_variables(lines_enum: ReqFileLines) -> ReqFileLines:
"""Replace all environment variables that can be retrieved via `os.getenv`.
The only allowed format for environment variables defined in the requirement file is `${MY_VARIABLE_1}` to
ensure two things:
1. Strings that contain a `$` aren't accidentally (partially) expanded.
2. Ensure consistency across platforms for requirement files.
These points are the result of a discussion on the `github pull request #3514
<https://github.com/pypa/pip/pull/3514>`_. Valid characters in variable names follow the `POSIX standard
<http://pubs.opengroup.org/onlinepubs/9699919799/>`_ and are limited to uppercase letter, digits and the `_`.
"""
for line_number, line in lines_enum:
for env_var, var_name in _ENV_VAR_RE.findall(line):
value = os.getenv(var_name)
if not value:
continue
line = line.replace(env_var, value)
yield line_number, line
@property
def as_root_args(self) -> list[str]:
if self._as_root_args is None:
opt = Namespace()
result: list[str] = []
for req in self._parse_requirements(opt=opt, recurse=False):
result.extend(req.as_args())
option_args = self._option_to_args(opt)
result.extend(option_args)
self._as_root_args = result
return self._as_root_args
def _option_to_args(self, opt: Namespace) -> list[str]:
result: list[str] = []
for req in getattr(opt, "requirements", []):
result.extend(("-r", req))
for req in getattr(opt, "constraints", []):
result.extend(("-c", req))
index_url = getattr(opt, "index_url", None)
if index_url is not None:
if index_url:
if index_url[0] != DEFAULT_INDEX_URL:
result.extend(("-i", index_url[0]))
for url in index_url[1:]:
result.extend(("--extra-index-url", url))
else:
result.append("--no-index")
for link in getattr(opt, "find_links", []):
result.extend(("-f", link))
if hasattr(opt, "pre"):
result.append("--pre")
for host in getattr(opt, "trusted_hosts", []):
result.extend(("--trusted-host", host))
if hasattr(opt, "prefer_binary"):
result.append("--prefer-binary")
if hasattr(opt, "require_hashes"):
result.append("--require-hashes")
for feature in getattr(opt, "features_enabled", []):
result.extend(("--use-feature", feature))
if hasattr(opt, "no_binary"):
result.extend(("--no-binary", opt.no_binary))
if hasattr(opt, "only_binary"):
result.extend(("--only-binary", opt.only_binary))
return result
__all__ = (
"RequirementsFile",
"ReqFileLines",
"ParsedRequirement",
)
|