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
1from __future__ import annotations
3import logging
4import os
5import shlex
6import subprocess
7from collections.abc import Callable, Iterable, Mapping
8from typing import Any, Literal
10from pip._vendor.rich.markup import escape
12from pip._internal.cli.spinners import SpinnerInterface, open_spinner
13from pip._internal.exceptions import InstallationSubprocessError
14from pip._internal.utils.logging import VERBOSE, subprocess_logger
15from pip._internal.utils.misc import HiddenText
17CommandArgs = list[str | HiddenText]
20def make_command(*args: str | HiddenText | CommandArgs) -> CommandArgs:
21 """
22 Create a CommandArgs object.
23 """
24 command_args: CommandArgs = []
25 for arg in args:
26 # Check for list instead of CommandArgs since CommandArgs is
27 # only known during type-checking.
28 if isinstance(arg, list):
29 command_args.extend(arg)
30 else:
31 # Otherwise, arg is str or HiddenText.
32 command_args.append(arg)
34 return command_args
37def format_command_args(args: list[str] | CommandArgs) -> str:
38 """
39 Format command arguments for display.
40 """
41 # For HiddenText arguments, display the redacted form by calling str().
42 # Also, we don't apply str() to arguments that aren't HiddenText since
43 # this can trigger a UnicodeDecodeError in Python 2 if the argument
44 # has type unicode and includes a non-ascii character. (The type
45 # checker doesn't ensure the annotations are correct in all cases.)
46 return " ".join(
47 shlex.quote(str(arg)) if isinstance(arg, HiddenText) else shlex.quote(arg)
48 for arg in args
49 )
52def reveal_command_args(args: list[str] | CommandArgs) -> list[str]:
53 """
54 Return the arguments in their raw, unredacted form.
55 """
56 return [arg.secret if isinstance(arg, HiddenText) else arg for arg in args]
59def call_subprocess(
60 cmd: list[str] | CommandArgs,
61 show_stdout: bool = False,
62 cwd: str | None = None,
63 on_returncode: Literal["raise", "warn", "ignore"] = "raise",
64 extra_ok_returncodes: Iterable[int] | None = None,
65 extra_environ: Mapping[str, Any] | None = None,
66 unset_environ: Iterable[str] | None = None,
67 spinner: SpinnerInterface | None = None,
68 log_failed_cmd: bool | None = True,
69 stdout_only: bool | None = False,
70 *,
71 command_desc: str,
72) -> str:
73 """
74 Args:
75 show_stdout: if true, use INFO to log the subprocess's stderr and
76 stdout streams. Otherwise, use DEBUG. Defaults to False.
77 extra_ok_returncodes: an iterable of integer return codes that are
78 acceptable, in addition to 0. Defaults to None, which means [].
79 unset_environ: an iterable of environment variable names to unset
80 prior to calling subprocess.Popen().
81 log_failed_cmd: if false, failed commands are not logged, only raised.
82 stdout_only: if true, return only stdout, else return both. When true,
83 logging of both stdout and stderr occurs when the subprocess has
84 terminated, else logging occurs as subprocess output is produced.
85 """
86 if extra_ok_returncodes is None:
87 extra_ok_returncodes = []
88 if unset_environ is None:
89 unset_environ = []
90 # Most places in pip use show_stdout=False. What this means is--
91 #
92 # - We connect the child's output (combined stderr and stdout) to a
93 # single pipe, which we read.
94 # - We log this output to stderr at DEBUG level as it is received.
95 # - If DEBUG logging isn't enabled (e.g. if --verbose logging wasn't
96 # requested), then we show a spinner so the user can still see the
97 # subprocess is in progress.
98 # - If the subprocess exits with an error, we log the output to stderr
99 # at ERROR level if it hasn't already been displayed to the console
100 # (e.g. if --verbose logging wasn't enabled). This way we don't log
101 # the output to the console twice.
102 #
103 # If show_stdout=True, then the above is still done, but with DEBUG
104 # replaced by INFO.
105 if show_stdout:
106 # Then log the subprocess output at INFO level.
107 log_subprocess: Callable[..., None] = subprocess_logger.info
108 used_level = logging.INFO
109 else:
110 # Then log the subprocess output using VERBOSE. This also ensures
111 # it will be logged to the log file (aka user_log), if enabled.
112 log_subprocess = subprocess_logger.verbose
113 used_level = VERBOSE
115 # Whether the subprocess will be visible in the console.
116 showing_subprocess = subprocess_logger.getEffectiveLevel() <= used_level
118 # Only use the spinner if we're not showing the subprocess output
119 # and we have a spinner.
120 use_spinner = not showing_subprocess and spinner is not None
122 log_subprocess("Running command %s", command_desc)
123 env = os.environ.copy()
124 if extra_environ:
125 env.update(extra_environ)
126 for name in unset_environ:
127 env.pop(name, None)
128 try:
129 proc = subprocess.Popen(
130 # Convert HiddenText objects to the underlying str.
131 reveal_command_args(cmd),
132 stdin=subprocess.PIPE,
133 stdout=subprocess.PIPE,
134 stderr=subprocess.STDOUT if not stdout_only else subprocess.PIPE,
135 cwd=cwd,
136 env=env,
137 encoding="locale",
138 errors="backslashreplace",
139 )
140 except Exception as exc:
141 if log_failed_cmd:
142 subprocess_logger.critical(
143 "Error %s while executing command %s",
144 exc,
145 command_desc,
146 )
147 raise
148 all_output = []
149 if not stdout_only:
150 assert proc.stdout
151 assert proc.stdin
152 proc.stdin.close()
153 # In this mode, stdout and stderr are in the same pipe.
154 while True:
155 line: str = proc.stdout.readline()
156 if not line:
157 break
158 line = line.rstrip()
159 all_output.append(line + "\n")
161 # Show the line immediately.
162 log_subprocess(line)
163 # Update the spinner.
164 if use_spinner:
165 assert spinner
166 spinner.spin()
167 try:
168 proc.wait()
169 finally:
170 if proc.stdout:
171 proc.stdout.close()
172 output = "".join(all_output)
173 else:
174 # In this mode, stdout and stderr are in different pipes.
175 # We must use communicate() which is the only safe way to read both.
176 out, err = proc.communicate()
177 # log line by line to preserve pip log indenting
178 for out_line in out.splitlines():
179 log_subprocess(out_line)
180 all_output.append(out)
181 for err_line in err.splitlines():
182 log_subprocess(err_line)
183 all_output.append(err)
184 output = out
186 proc_had_error = proc.returncode and proc.returncode not in extra_ok_returncodes
187 if use_spinner:
188 assert spinner
189 if proc_had_error:
190 spinner.finish("error")
191 else:
192 spinner.finish("done")
193 if proc_had_error:
194 if on_returncode == "raise":
195 error = InstallationSubprocessError(
196 command_description=command_desc,
197 exit_code=proc.returncode,
198 output_lines=all_output if not showing_subprocess else None,
199 )
200 if log_failed_cmd:
201 subprocess_logger.error("%s", error, extra={"rich": True})
202 subprocess_logger.verbose(
203 "[bold magenta]full command[/]: [blue]%s[/]",
204 escape(format_command_args(cmd)),
205 extra={"markup": True},
206 )
207 subprocess_logger.verbose(
208 "[bold magenta]cwd[/]: %s",
209 escape(cwd or "[inherit]"),
210 extra={"markup": True},
211 )
213 raise error
214 elif on_returncode == "warn":
215 subprocess_logger.warning(
216 'Command "%s" had error code %s in %s',
217 command_desc,
218 proc.returncode,
219 cwd,
220 )
221 elif on_returncode == "ignore":
222 pass
223 else:
224 raise ValueError(f"Invalid value: on_returncode={on_returncode!r}")
225 return output
228def runner_with_spinner_message(message: str) -> Callable[..., None]:
229 """Provide a subprocess_runner that shows a spinner message.
231 Intended for use with for BuildBackendHookCaller. Thus, the runner has
232 an API that matches what's expected by BuildBackendHookCaller.subprocess_runner.
233 """
235 def runner(
236 cmd: list[str],
237 cwd: str | None = None,
238 extra_environ: Mapping[str, Any] | None = None,
239 ) -> None:
240 with open_spinner(message) as spinner:
241 call_subprocess(
242 cmd,
243 command_desc=message,
244 cwd=cwd,
245 extra_environ=extra_environ,
246 spinner=spinner,
247 )
249 return runner