Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2Requirements file parsing
3"""
5from __future__ import annotations
7import codecs
8import logging
9import optparse
10import os
11import re
12import shlex
13import sys
14import urllib.parse
15from collections.abc import Callable, Generator, Iterable
16from dataclasses import dataclass
17from optparse import Values
18from typing import (
19 TYPE_CHECKING,
20 Any,
21 NoReturn,
22)
24from pip._internal.cli import cmdoptions
25from pip._internal.exceptions import InstallationError, RequirementsFileParseError
26from pip._internal.models.release_control import ReleaseControl
27from pip._internal.models.search_scope import SearchScope
28from pip._internal.utils.compat import get_locale_encoding
30if TYPE_CHECKING:
31 from pip._internal.index.package_finder import PackageFinder
32 from pip._internal.network.session import PipSession
34__all__ = ["parse_requirements"]
36ReqFileLines = Iterable[tuple[int, str]]
38LineParser = Callable[[str], tuple[str, Values]]
40SCHEME_RE = re.compile(r"^(http|https|file):", re.I)
41COMMENT_RE = re.compile(r"(^|\s+)#.*$")
43# Matches environment variable-style values in '${MY_VARIABLE_1}' with the
44# variable name consisting of only uppercase letters, digits or the '_'
45# (underscore). This follows the POSIX standard defined in IEEE Std 1003.1,
46# 2013 Edition.
47ENV_VAR_RE = re.compile(r"(?P<var>\$\{(?P<name>[A-Z0-9_]+)\})")
49SUPPORTED_OPTIONS: list[Callable[..., optparse.Option]] = [
50 cmdoptions.index_url,
51 cmdoptions.extra_index_url,
52 cmdoptions.no_index,
53 cmdoptions.constraints,
54 cmdoptions.requirements,
55 cmdoptions.editable,
56 cmdoptions.find_links,
57 cmdoptions.no_binary,
58 cmdoptions.only_binary,
59 cmdoptions.prefer_binary,
60 cmdoptions.require_hashes,
61 cmdoptions.no_require_hashes,
62 cmdoptions.pre,
63 cmdoptions.all_releases,
64 cmdoptions.only_final,
65 cmdoptions.trusted_host,
66 cmdoptions.use_new_feature,
67]
69# options to be passed to requirements
70SUPPORTED_OPTIONS_REQ: list[Callable[..., optparse.Option]] = [
71 cmdoptions.hash,
72 cmdoptions.config_settings,
73]
75SUPPORTED_OPTIONS_EDITABLE_REQ: list[Callable[..., optparse.Option]] = [
76 cmdoptions.config_settings,
77]
80# the 'dest' string values
81SUPPORTED_OPTIONS_REQ_DEST = [str(o().dest) for o in SUPPORTED_OPTIONS_REQ]
82SUPPORTED_OPTIONS_EDITABLE_REQ_DEST = [
83 str(o().dest) for o in SUPPORTED_OPTIONS_EDITABLE_REQ
84]
86# order of BOMS is important: codecs.BOM_UTF16_LE is a prefix of codecs.BOM_UTF32_LE
87# so data.startswith(BOM_UTF16_LE) would be true for UTF32_LE data
88BOMS: list[tuple[bytes, str]] = [
89 (codecs.BOM_UTF8, "utf-8"),
90 (codecs.BOM_UTF32, "utf-32"),
91 (codecs.BOM_UTF32_BE, "utf-32-be"),
92 (codecs.BOM_UTF32_LE, "utf-32-le"),
93 (codecs.BOM_UTF16, "utf-16"),
94 (codecs.BOM_UTF16_BE, "utf-16-be"),
95 (codecs.BOM_UTF16_LE, "utf-16-le"),
96]
98PEP263_ENCODING_RE = re.compile(rb"coding[:=]\s*([-\w.]+)")
99DEFAULT_ENCODING = "utf-8"
101logger = logging.getLogger(__name__)
104@dataclass(frozen=True, slots=True)
105class ParsedRequirement:
106 requirement: str
107 is_editable: bool
108 comes_from: str
109 constraint: bool
110 options: dict[str, Any] | None
111 line_source: str | None
114@dataclass(frozen=True, slots=True)
115class ParsedLine:
116 filename: str
117 lineno: int
118 args: str
119 opts: Values
120 constraint: bool
122 @property
123 def is_editable(self) -> bool:
124 return bool(self.opts.editables)
126 @property
127 def requirement(self) -> str | None:
128 if self.args:
129 return self.args
130 elif self.is_editable:
131 # We don't support multiple -e on one line
132 return self.opts.editables[0]
133 return None
136def parse_requirements(
137 filename: str,
138 session: PipSession,
139 finder: PackageFinder | None = None,
140 options: optparse.Values | None = None,
141 constraint: bool = False,
142) -> Generator[ParsedRequirement, None, None]:
143 """Parse a requirements file and yield ParsedRequirement instances.
145 :param filename: Path or url of requirements file.
146 :param session: PipSession instance.
147 :param finder: Instance of pip.index.PackageFinder.
148 :param options: cli options.
149 :param constraint: If true, parsing a constraint file rather than
150 requirements file.
151 """
152 line_parser = get_line_parser(finder)
153 parser = RequirementsFileParser(session, line_parser)
155 for parsed_line in parser.parse(filename, constraint):
156 parsed_req = handle_line(
157 parsed_line, options=options, finder=finder, session=session
158 )
159 if parsed_req is not None:
160 yield parsed_req
163def preprocess(content: str) -> ReqFileLines:
164 """Split, filter, and join lines, and return a line iterator
166 :param content: the content of the requirements file
167 """
168 lines_enum: ReqFileLines = enumerate(content.splitlines(), start=1)
169 lines_enum = join_lines(lines_enum)
170 lines_enum = ignore_comments(lines_enum)
171 lines_enum = expand_env_variables(lines_enum)
172 return lines_enum
175def handle_requirement_line(
176 line: ParsedLine,
177 options: optparse.Values | None = None,
178) -> ParsedRequirement:
179 # preserve for the nested code path
180 line_comes_from = "{} {} (line {})".format(
181 "-c" if line.constraint else "-r",
182 line.filename,
183 line.lineno,
184 )
186 assert line.requirement is not None
188 # get the options that apply to requirements
189 if line.is_editable:
190 supported_dest = SUPPORTED_OPTIONS_EDITABLE_REQ_DEST
191 else:
192 supported_dest = SUPPORTED_OPTIONS_REQ_DEST
193 req_options = {}
194 for dest in supported_dest:
195 if dest in line.opts.__dict__ and line.opts.__dict__[dest]:
196 req_options[dest] = line.opts.__dict__[dest]
198 line_source = f"line {line.lineno} of {line.filename}"
199 return ParsedRequirement(
200 requirement=line.requirement,
201 is_editable=line.is_editable,
202 comes_from=line_comes_from,
203 constraint=line.constraint,
204 options=req_options,
205 line_source=line_source,
206 )
209def handle_option_line(
210 opts: Values,
211 filename: str,
212 lineno: int,
213 finder: PackageFinder | None = None,
214 options: optparse.Values | None = None,
215 session: PipSession | None = None,
216) -> None:
217 if opts.hashes:
218 logger.warning(
219 "%s line %s has --hash but no requirement, and will be ignored.",
220 filename,
221 lineno,
222 )
224 if options:
225 # percolate options upward
226 if opts.require_hashes:
227 options.require_hashes = opts.require_hashes
228 if opts.no_require_hashes:
229 options.no_require_hashes = opts.no_require_hashes
230 if opts.features_enabled:
231 options.features_enabled.extend(
232 f for f in opts.features_enabled if f not in options.features_enabled
233 )
235 # set finder options
236 if finder:
237 find_links = finder.find_links
238 index_urls = finder.index_urls
239 no_index = finder.search_scope.no_index
240 if opts.no_index is True:
241 no_index = True
242 index_urls = []
243 if opts.index_url and not no_index:
244 index_urls = [opts.index_url]
245 if opts.extra_index_urls and not no_index:
246 index_urls.extend(opts.extra_index_urls)
247 if opts.find_links:
248 # FIXME: it would be nice to keep track of the source
249 # of the find_links: support a find-links local path
250 # relative to a requirements file.
251 value = opts.find_links[0]
252 req_dir = os.path.dirname(os.path.abspath(filename))
253 relative_to_reqs_file = os.path.join(req_dir, value)
254 if os.path.exists(relative_to_reqs_file):
255 value = relative_to_reqs_file
256 find_links.append(value)
258 if session:
259 # We need to update the auth urls in session
260 session.update_index_urls(index_urls)
262 search_scope = SearchScope(
263 find_links=find_links,
264 index_urls=index_urls,
265 no_index=no_index,
266 )
267 finder.search_scope = search_scope
269 # Transform --pre into --all-releases :all:
270 if opts.pre:
271 if not opts.release_control:
272 opts.release_control = ReleaseControl()
273 opts.release_control.all_releases.add(":all:")
275 if opts.release_control:
276 if not finder.release_control:
277 # First time seeing release_control, set it on finder
278 finder.set_release_control(opts.release_control)
280 if opts.prefer_binary:
281 finder.set_prefer_binary()
283 if session:
284 for host in opts.trusted_hosts or []:
285 source = f"line {lineno} of {filename}"
286 session.add_trusted_host(host, source=source)
289def handle_line(
290 line: ParsedLine,
291 options: optparse.Values | None = None,
292 finder: PackageFinder | None = None,
293 session: PipSession | None = None,
294) -> ParsedRequirement | None:
295 """Handle a single parsed requirements line; This can result in
296 creating/yielding requirements, or updating the finder.
298 :param line: The parsed line to be processed.
299 :param options: CLI options.
300 :param finder: The finder - updated by non-requirement lines.
301 :param session: The session - updated by non-requirement lines.
303 Returns a ParsedRequirement object if the line is a requirement line,
304 otherwise returns None.
306 For lines that contain requirements, the only options that have an effect
307 are from SUPPORTED_OPTIONS_REQ, and they are scoped to the
308 requirement. Other options from SUPPORTED_OPTIONS may be present, but are
309 ignored.
311 For lines that do not contain requirements, the only options that have an
312 effect are from SUPPORTED_OPTIONS. Options from SUPPORTED_OPTIONS_REQ may
313 be present, but are ignored. These lines may contain multiple options
314 (although our docs imply only one is supported), and all our parsed and
315 affect the finder.
316 """
318 if line.requirement is not None:
319 parsed_req = handle_requirement_line(line, options)
320 return parsed_req
321 else:
322 handle_option_line(
323 line.opts,
324 line.filename,
325 line.lineno,
326 finder,
327 options,
328 session,
329 )
330 return None
333class RequirementsFileParser:
334 def __init__(
335 self,
336 session: PipSession,
337 line_parser: LineParser,
338 ) -> None:
339 self._session = session
340 self._line_parser = line_parser
342 def parse(
343 self, filename: str, constraint: bool
344 ) -> Generator[ParsedLine, None, None]:
345 """Parse a given file, yielding parsed lines."""
346 yield from self._parse_and_recurse(
347 filename, constraint, [{os.path.abspath(filename): None}]
348 )
350 def _parse_and_recurse(
351 self,
352 filename: str,
353 constraint: bool,
354 parsed_files_stack: list[dict[str, str | None]],
355 ) -> Generator[ParsedLine, None, None]:
356 for line in self._parse_file(filename, constraint):
357 if line.requirement is None and (
358 line.opts.requirements or line.opts.constraints
359 ):
360 # parse a nested requirements file
361 if line.opts.requirements:
362 req_path = line.opts.requirements[0]
363 nested_constraint = False
364 else:
365 req_path = line.opts.constraints[0]
366 nested_constraint = True
368 # original file is over http
369 if SCHEME_RE.search(filename):
370 # do a url join so relative paths work
371 req_path = urllib.parse.urljoin(filename, req_path)
372 # original file and nested file are paths
373 elif not SCHEME_RE.search(req_path):
374 # do a join so relative paths work
375 # and then abspath so that we can identify recursive references
376 req_path = os.path.abspath(
377 os.path.join(
378 os.path.dirname(filename),
379 req_path,
380 )
381 )
382 parsed_files = parsed_files_stack[0]
383 if req_path in parsed_files:
384 initial_file = parsed_files[req_path]
385 tail = (
386 f" and again in {initial_file}"
387 if initial_file is not None
388 else ""
389 )
390 raise RequirementsFileParseError(
391 f"{req_path} recursively references itself in {filename}{tail}"
392 )
393 # Keeping a track where was each file first included in
394 new_parsed_files = parsed_files.copy()
395 new_parsed_files[req_path] = filename
396 yield from self._parse_and_recurse(
397 req_path, nested_constraint, [new_parsed_files, *parsed_files_stack]
398 )
399 else:
400 yield line
402 def _parse_file(
403 self, filename: str, constraint: bool
404 ) -> Generator[ParsedLine, None, None]:
405 _, content = get_file_content(filename, self._session, constraint=constraint)
407 lines_enum = preprocess(content)
409 for line_number, line in lines_enum:
410 try:
411 args_str, opts = self._line_parser(line)
412 except OptionParsingError as e:
413 # add offending line
414 msg = f"Invalid requirement: {line}\n{e.msg}"
415 raise RequirementsFileParseError(msg)
417 yield ParsedLine(
418 filename,
419 line_number,
420 args_str,
421 opts,
422 constraint,
423 )
426def get_line_parser(finder: PackageFinder | None) -> LineParser:
427 def parse_line(line: str) -> tuple[str, Values]:
428 # Build new parser for each line since it accumulates appendable
429 # options.
430 parser = build_parser()
431 defaults = parser.get_default_values()
432 defaults.index_url = None
433 if finder:
434 defaults.format_control = finder.format_control
435 defaults.release_control = finder.release_control
437 args_str, options_str = break_args_options(line)
439 try:
440 options = shlex.split(options_str)
441 except ValueError as e:
442 raise OptionParsingError(f"Could not split options: {options_str}") from e
444 opts, _ = parser.parse_args(options, defaults)
446 return args_str, opts
448 return parse_line
451def break_args_options(line: str) -> tuple[str, str]:
452 """Break up the line into an args and options string. We only want to shlex
453 (and then optparse) the options, not the args. args can contain markers
454 which are corrupted by shlex.
455 """
456 tokens = line.split(" ")
457 args = []
458 options = tokens[:]
459 for token in tokens:
460 if token.startswith(("-", "--")):
461 break
462 else:
463 args.append(token)
464 options.pop(0)
465 return " ".join(args), " ".join(options)
468class OptionParsingError(Exception):
469 def __init__(self, msg: str) -> None:
470 self.msg = msg
473def build_parser() -> optparse.OptionParser:
474 """
475 Return a parser for parsing requirement lines
476 """
477 parser = optparse.OptionParser(add_help_option=False)
479 option_factories = SUPPORTED_OPTIONS + SUPPORTED_OPTIONS_REQ
480 for option_factory in option_factories:
481 option = option_factory()
482 parser.add_option(option)
484 # By default optparse sys.exits on parsing errors. We want to wrap
485 # that in our own exception.
486 def parser_exit(self: Any, msg: str) -> NoReturn:
487 raise OptionParsingError(msg)
489 # NOTE: mypy disallows assigning to a method
490 # https://github.com/python/mypy/issues/2427
491 parser.exit = parser_exit # type: ignore
493 return parser
496def join_lines(lines_enum: ReqFileLines) -> ReqFileLines:
497 """Joins a line ending in '\' with the previous line (except when following
498 comments). The joined line takes on the index of the first line.
499 """
500 primary_line_number = None
501 new_line: list[str] = []
502 for line_number, line in lines_enum:
503 if not line.endswith("\\") or COMMENT_RE.match(line):
504 if COMMENT_RE.match(line):
505 # this ensures comments are always matched later
506 line = " " + line
507 if new_line:
508 new_line.append(line)
509 assert primary_line_number is not None
510 yield primary_line_number, "".join(new_line)
511 new_line = []
512 else:
513 yield line_number, line
514 else:
515 if not new_line:
516 primary_line_number = line_number
517 new_line.append(line.strip("\\"))
519 # last line contains \
520 if new_line:
521 assert primary_line_number is not None
522 yield primary_line_number, "".join(new_line)
525def ignore_comments(lines_enum: ReqFileLines) -> ReqFileLines:
526 """
527 Strips comments and filter empty lines.
528 """
529 for line_number, line in lines_enum:
530 line = COMMENT_RE.sub("", line)
531 line = line.strip()
532 if line:
533 yield line_number, line
536def expand_env_variables(lines_enum: ReqFileLines) -> ReqFileLines:
537 """Replace all environment variables that can be retrieved via `os.getenv`.
539 The only allowed format for environment variables defined in the
540 requirement file is `${MY_VARIABLE_1}` to ensure two things:
542 1. Strings that contain a `$` aren't accidentally (partially) expanded.
543 2. Ensure consistency across platforms for requirement files.
545 These points are the result of a discussion on the `github pull
546 request #3514 <https://github.com/pypa/pip/pull/3514>`_.
548 Valid characters in variable names follow the `POSIX standard
549 <http://pubs.opengroup.org/onlinepubs/9699919799/>`_ and are limited
550 to uppercase letter, digits and the `_` (underscore).
551 """
552 for line_number, line in lines_enum:
553 for env_var, var_name in ENV_VAR_RE.findall(line):
554 value = os.getenv(var_name)
555 if not value:
556 continue
558 line = line.replace(env_var, value)
560 yield line_number, line
563def get_file_content(
564 url: str, session: PipSession, *, constraint: bool = False
565) -> tuple[str, str]:
566 """Gets the content of a file; it may be a filename, file: URL, or
567 http: URL. Returns (location, content). Content is unicode.
568 Respects # -*- coding: declarations on the retrieved files.
570 :param url: File path or url.
571 :param session: PipSession instance.
572 """
573 scheme = urllib.parse.urlsplit(url).scheme
574 # Pip has special support for file:// URLs (LocalFSAdapter).
575 if scheme in ["http", "https", "file"]:
576 # Delay importing heavy network modules until absolutely necessary.
577 from pip._internal.network.utils import raise_for_status
579 resp = session.get(url)
580 raise_for_status(resp)
581 return resp.url, resp.text
583 # Assume this is a bare path.
584 try:
585 with open(url, "rb") as f:
586 raw_content = f.read()
587 except OSError as exc:
588 kind = "constraint" if constraint else "requirements"
589 raise InstallationError(f"Could not open {kind} file: {exc}")
591 content = _decode_req_file(raw_content, url)
593 return url, content
596def _decode_req_file(data: bytes, url: str) -> str:
597 for bom, encoding in BOMS:
598 if data.startswith(bom):
599 return data[len(bom) :].decode(encoding)
601 for line in data.split(b"\n")[:2]:
602 if line[0:1] == b"#":
603 result = PEP263_ENCODING_RE.search(line)
604 if result is not None:
605 encoding = result.groups()[0].decode("ascii")
606 return data.decode(encoding)
608 try:
609 return data.decode(DEFAULT_ENCODING)
610 except UnicodeDecodeError:
611 locale_encoding = get_locale_encoding() or sys.getdefaultencoding()
612 logging.warning(
613 "unable to decode data from %s with default encoding %s, "
614 "falling back to encoding from locale: %s. "
615 "If this is intentional you should specify the encoding with a "
616 "PEP-263 style comment, e.g. '# -*- coding: %s -*-'",
617 url,
618 DEFAULT_ENCODING,
619 locale_encoding,
620 locale_encoding,
621 )
622 return data.decode(locale_encoding)