diff options
Diffstat (limited to 'src/pip/_internal/utils/subprocess.py')
| -rw-r--r-- | src/pip/_internal/utils/subprocess.py | 143 |
1 files changed, 61 insertions, 82 deletions
diff --git a/src/pip/_internal/utils/subprocess.py b/src/pip/_internal/utils/subprocess.py index da052ee69..cf5bf6be1 100644 --- a/src/pip/_internal/utils/subprocess.py +++ b/src/pip/_internal/utils/subprocess.py @@ -2,25 +2,38 @@ import logging import os import shlex import subprocess -from typing import Any, Callable, Iterable, List, Mapping, Optional, Union +from typing import ( + TYPE_CHECKING, + Any, + Callable, + Iterable, + List, + Mapping, + Optional, + Union, +) + +from pip._vendor.rich.markup import escape from pip._internal.cli.spinners import SpinnerInterface, open_spinner from pip._internal.exceptions import InstallationSubprocessError from pip._internal.utils.logging import VERBOSE, subprocess_logger from pip._internal.utils.misc import HiddenText -CommandArgs = List[Union[str, HiddenText]] - +if TYPE_CHECKING: + # Literal was introduced in Python 3.8. + # + # TODO: Remove `if TYPE_CHECKING` when dropping support for Python 3.7. + from typing import Literal -LOG_DIVIDER = "----------------------------------------" +CommandArgs = List[Union[str, HiddenText]] -def make_command(*args): - # type: (Union[str, HiddenText, CommandArgs]) -> CommandArgs +def make_command(*args: Union[str, HiddenText, CommandArgs]) -> CommandArgs: """ Create a CommandArgs object. """ - command_args = [] # type: CommandArgs + command_args: CommandArgs = [] for arg in args: # Check for list instead of CommandArgs since CommandArgs is # only known during type-checking. @@ -33,8 +46,7 @@ def make_command(*args): return command_args -def format_command_args(args): - # type: (Union[List[str], CommandArgs]) -> str +def format_command_args(args: Union[List[str], CommandArgs]) -> str: """ Format command arguments for display. """ @@ -49,64 +61,27 @@ def format_command_args(args): ) -def reveal_command_args(args): - # type: (Union[List[str], CommandArgs]) -> List[str] +def reveal_command_args(args: Union[List[str], CommandArgs]) -> List[str]: """ Return the arguments in their raw, unredacted form. """ return [arg.secret if isinstance(arg, HiddenText) else arg for arg in args] -def make_subprocess_output_error( - cmd_args, # type: Union[List[str], CommandArgs] - cwd, # type: Optional[str] - lines, # type: List[str] - exit_status, # type: int -): - # type: (...) -> str - """ - Create and return the error message to use to log a subprocess error - with command output. - - :param lines: A list of lines, each ending with a newline. - """ - command = format_command_args(cmd_args) - - # We know the joined output value ends in a newline. - output = "".join(lines) - msg = ( - # Use a unicode string to avoid "UnicodeEncodeError: 'ascii' - # codec can't encode character ..." in Python 2 when a format - # argument (e.g. `output`) has a non-ascii character. - "Command errored out with exit status {exit_status}:\n" - " command: {command_display}\n" - " cwd: {cwd_display}\n" - "Complete output ({line_count} lines):\n{output}{divider}" - ).format( - exit_status=exit_status, - command_display=command, - cwd_display=cwd, - line_count=len(lines), - output=output, - divider=LOG_DIVIDER, - ) - return msg - - def call_subprocess( - cmd, # type: Union[List[str], CommandArgs] - show_stdout=False, # type: bool - cwd=None, # type: Optional[str] - on_returncode="raise", # type: str - extra_ok_returncodes=None, # type: Optional[Iterable[int]] - command_desc=None, # type: Optional[str] - extra_environ=None, # type: Optional[Mapping[str, Any]] - unset_environ=None, # type: Optional[Iterable[str]] - spinner=None, # type: Optional[SpinnerInterface] - log_failed_cmd=True, # type: Optional[bool] - stdout_only=False, # type: Optional[bool] -): - # type: (...) -> str + cmd: Union[List[str], CommandArgs], + show_stdout: bool = False, + cwd: Optional[str] = None, + on_returncode: 'Literal["raise", "warn", "ignore"]' = "raise", + extra_ok_returncodes: Optional[Iterable[int]] = None, + extra_environ: Optional[Mapping[str, Any]] = None, + unset_environ: Optional[Iterable[str]] = None, + spinner: Optional[SpinnerInterface] = None, + log_failed_cmd: Optional[bool] = True, + stdout_only: Optional[bool] = False, + *, + command_desc: str, +) -> str: """ Args: show_stdout: if true, use INFO to log the subprocess's stderr and @@ -141,7 +116,7 @@ def call_subprocess( # replaced by INFO. if show_stdout: # Then log the subprocess output at INFO level. - log_subprocess = subprocess_logger.info + log_subprocess: Callable[..., None] = subprocess_logger.info used_level = logging.INFO else: # Then log the subprocess output using VERBOSE. This also ensures @@ -156,9 +131,6 @@ def call_subprocess( # and we have a spinner. use_spinner = not showing_subprocess and spinner is not None - if command_desc is None: - command_desc = format_command_args(cmd) - log_subprocess("Running command %s", command_desc) env = os.environ.copy() if extra_environ: @@ -191,7 +163,7 @@ def call_subprocess( proc.stdin.close() # In this mode, stdout and stderr are in the same pipe. while True: - line = proc.stdout.readline() # type: str + line: str = proc.stdout.readline() if not line: break line = line.rstrip() @@ -231,17 +203,25 @@ def call_subprocess( spinner.finish("done") if proc_had_error: if on_returncode == "raise": - if not showing_subprocess and log_failed_cmd: - # Then the subprocess streams haven't been logged to the - # console yet. - msg = make_subprocess_output_error( - cmd_args=cmd, - cwd=cwd, - lines=all_output, - exit_status=proc.returncode, + error = InstallationSubprocessError( + command_description=command_desc, + exit_code=proc.returncode, + output_lines=all_output if not showing_subprocess else None, + ) + if log_failed_cmd: + subprocess_logger.error("[present-rich] %s", error) + subprocess_logger.verbose( + "[bold magenta]full command[/]: [blue]%s[/]", + escape(format_command_args(cmd)), + extra={"markup": True}, ) - subprocess_logger.error(msg) - raise InstallationSubprocessError(proc.returncode, command_desc) + subprocess_logger.verbose( + "[bold magenta]cwd[/]: %s", + escape(cwd or "[inherit]"), + extra={"markup": True}, + ) + + raise error elif on_returncode == "warn": subprocess_logger.warning( 'Command "%s" had error code %s in %s', @@ -256,8 +236,7 @@ def call_subprocess( return output -def runner_with_spinner_message(message): - # type: (str) -> Callable[..., None] +def runner_with_spinner_message(message: str) -> Callable[..., None]: """Provide a subprocess_runner that shows a spinner message. Intended for use with for pep517's Pep517HookCaller. Thus, the runner has @@ -265,14 +244,14 @@ def runner_with_spinner_message(message): """ def runner( - cmd, # type: List[str] - cwd=None, # type: Optional[str] - extra_environ=None, # type: Optional[Mapping[str, Any]] - ): - # type: (...) -> None + cmd: List[str], + cwd: Optional[str] = None, + extra_environ: Optional[Mapping[str, Any]] = None, + ) -> None: with open_spinner(message) as spinner: call_subprocess( cmd, + command_desc=message, cwd=cwd, extra_environ=extra_environ, spinner=spinner, |