FM / utils

/usr/lib/python3.8/site-packages/pip/_internal/utils/

/usr/lib/python3.8/site-packages/pip/_internal/utils

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
appdirs.py	9766	0644	edit dl rm
compat.py	9565	0644	edit dl rm
deprecation.py	3318	0644	edit dl rm
encoding.py	1320	0644	edit dl rm
filesystem.py	3334	0644	edit dl rm
filetypes.py	571	0644	edit dl rm
glibc.py	4397	0644	edit dl rm
hashes.py	4020	0644	edit dl rm
inject_securetransport.py	810	0644	edit dl rm
logging.py	13033	0644	edit dl rm
marker_files.py	823	0644	edit dl rm
misc.py	25579	0644	edit dl rm
models.py	1148	0644	edit dl rm
packaging.py	3035	0644	edit dl rm
setuptools_build.py	1631	0644	edit dl rm
subprocess.py	9911	0644	edit dl rm
temp_dir.py	5521	0644	edit dl rm
typing.py	1125	0644	edit dl rm
ui.py	13906	0644	edit dl rm
unpacking.py	9682	0644	edit dl rm
urls.py	1481	0644	edit dl rm
virtualenv.py	891	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /usr/lib/python3.8/site-packages/pip/_internal/utils/subprocess.py (9911B)

# The following comment should be removed at some point in the future.
# mypy: strict-optional=False

from __future__ import absolute_import

import logging
import os
import subprocess

from pip._vendor.six.moves import shlex_quote

from pip._internal.exceptions import InstallationError
from pip._internal.utils.compat import console_to_str, str_to_display
from pip._internal.utils.logging import subprocess_logger
from pip._internal.utils.misc import HiddenText, path_to_display
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
from pip._internal.utils.ui import open_spinner

if MYPY_CHECK_RUNNING:
    from typing import (
        Any, Callable, Iterable, List, Mapping, Optional, Text, Union,
    )
    from pip._internal.utils.ui import SpinnerInterface

CommandArgs = List[Union[str, HiddenText]]

LOG_DIVIDER = '----------------------------------------'

def make_command(*args):
    # type: (Union[str, HiddenText, CommandArgs]) -> CommandArgs
    """
    Create a CommandArgs object.
    """
    command_args = []  # type: CommandArgs
    for arg in args:
        # Check for list instead of CommandArgs since CommandArgs is
        # only known during type-checking.
        if isinstance(arg, list):
            command_args.extend(arg)
        else:
            # Otherwise, arg is str or HiddenText.
            command_args.append(arg)

return command_args

def format_command_args(args):
    # type: (Union[List[str], CommandArgs]) -> str
    """
    Format command arguments for display.
    """
    # For HiddenText arguments, display the redacted form by calling str().
    # Also, we don't apply str() to arguments that aren't HiddenText since
    # this can trigger a UnicodeDecodeError in Python 2 if the argument
    # has type unicode and includes a non-ascii character.  (The type
    # checker doesn't ensure the annotations are correct in all cases.)
    return ' '.join(
        shlex_quote(str(arg)) if isinstance(arg, HiddenText)
        else shlex_quote(arg) for arg in args
    )

def reveal_command_args(args):
    # type: (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[Text]
    exit_status,  # type: int
):
    # type: (...) -> Text
    """
    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)
    # Convert `command` and `cwd` to text (unicode in Python 2) so we can use
    # them as arguments in the unicode format string below. This avoids
    # "UnicodeDecodeError: 'ascii' codec can't decode byte ..." in Python 2
    # if either contains a non-ascii character.
    command_display = str_to_display(command, desc='command bytes')
    cwd_display = path_to_display(cwd)

# 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.
        u'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_display,
        cwd_display=cwd_display,
        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]
):
    # type: (...) -> Text
    """
    Args:
      show_stdout: if true, use INFO to log the subprocess's stderr and
        stdout streams.  Otherwise, use DEBUG.  Defaults to False.
      extra_ok_returncodes: an iterable of integer return codes that are
        acceptable, in addition to 0. Defaults to None, which means [].
      unset_environ: an iterable of environment variable names to unset
        prior to calling subprocess.Popen().
      log_failed_cmd: if false, failed commands are not logged, only raised.
    """
    if extra_ok_returncodes is None:
        extra_ok_returncodes = []
    if unset_environ is None:
        unset_environ = []
    # Most places in pip use show_stdout=False. What this means is--
    #
    # - We connect the child's output (combined stderr and stdout) to a
    #   single pipe, which we read.
    # - We log this output to stderr at DEBUG level as it is received.
    # - If DEBUG logging isn't enabled (e.g. if --verbose logging wasn't
    #   requested), then we show a spinner so the user can still see the
    #   subprocess is in progress.
    # - If the subprocess exits with an error, we log the output to stderr
    #   at ERROR level if it hasn't already been displayed to the console
    #   (e.g. if --verbose logging wasn't enabled).  This way we don't log
    #   the output to the console twice.
    #
    # If show_stdout=True, then the above is still done, but with DEBUG
    # replaced by INFO.
    if show_stdout:
        # Then log the subprocess output at INFO level.
        log_subprocess = subprocess_logger.info
        used_level = logging.INFO
    else:
        # Then log the subprocess output using DEBUG.  This also ensures
        # it will be logged to the log file (aka user_log), if enabled.
        log_subprocess = subprocess_logger.debug
        used_level = logging.DEBUG

# Whether the subprocess will be visible in the console.
    showing_subprocess = subprocess_logger.getEffectiveLevel() <= used_level

# Only use the spinner if we're not showing the subprocess output
    # 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:
        env.update(extra_environ)
    for name in unset_environ:
        env.pop(name, None)
    try:
        proc = subprocess.Popen(
            # Convert HiddenText objects to the underlying str.
            reveal_command_args(cmd),
            stderr=subprocess.STDOUT, stdin=subprocess.PIPE,
            stdout=subprocess.PIPE, cwd=cwd, env=env,
        )
        proc.stdin.close()
    except Exception as exc:
        if log_failed_cmd:
            subprocess_logger.critical(
                "Error %s while executing command %s", exc, command_desc,
            )
        raise
    all_output = []
    while True:
        # The "line" value is a unicode string in Python 2.
        line = console_to_str(proc.stdout.readline())
        if not line:
            break
        line = line.rstrip()
        all_output.append(line + '\n')

# Show the line immediately.
        log_subprocess(line)
        # Update the spinner.
        if use_spinner:
            spinner.spin()
    try:
        proc.wait()
    finally:
        if proc.stdout:
            proc.stdout.close()
    proc_had_error = (
        proc.returncode and proc.returncode not in extra_ok_returncodes
    )
    if use_spinner:
        if proc_had_error:
            spinner.finish("error")
        else:
            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,
                )
                subprocess_logger.error(msg)
            exc_msg = (
                'Command errored out with exit status {}: {} '
                'Check the logs for full command output.'
            ).format(proc.returncode, command_desc)
            raise InstallationError(exc_msg)
        elif on_returncode == 'warn':
            subprocess_logger.warning(
                'Command "%s" had error code %s in %s',
                command_desc, proc.returncode, cwd,
            )
        elif on_returncode == 'ignore':
            pass
        else:
            raise ValueError('Invalid value: on_returncode=%s' %
                             repr(on_returncode))
    return ''.join(all_output)

def runner_with_spinner_message(message):
    # type: (str) -> Callable
    """Provide a subprocess_runner that shows a spinner message.

Intended for use with for pep517's Pep517HookCaller. Thus, the runner has
    an API that matches what's expected by Pep517HookCaller.subprocess_runner.
    """

def runner(
        cmd,  # type: List[str]
        cwd=None,  # type: Optional[str]
        extra_environ=None  # type: Optional[Mapping[str, Any]]
    ):
        # type: (...) -> None
        with open_spinner(message) as spinner:
            call_subprocess(
                cmd,
                cwd=cwd,
                extra_environ=extra_environ,
                spinner=spinner,
            )

return runner

cmd: