base_command.py 7.84 KB
Newer Older
xuebingbing's avatar
xuebingbing committed
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
"""Base Command class, and related routines"""

from __future__ import absolute_import, print_function

import logging
import logging.config
import optparse
import os
import platform
import sys
import traceback

from pip._internal.cli import cmdoptions
from pip._internal.cli.command_context import CommandContextMixIn
from pip._internal.cli.parser import (
    ConfigOptionParser,
    UpdatingDefaultsHelpFormatter,
)
from pip._internal.cli.status_codes import (
    ERROR,
    PREVIOUS_BUILD_DIR_ERROR,
    SUCCESS,
    UNKNOWN_ERROR,
    VIRTUALENV_NOT_FOUND,
)
from pip._internal.exceptions import (
    BadCommand,
    CommandError,
    InstallationError,
    PreviousBuildDirError,
    UninstallationError,
)
from pip._internal.utils.deprecation import deprecated
from pip._internal.utils.filesystem import check_path_owner
from pip._internal.utils.logging import BrokenStdoutLoggingError, setup_logging
from pip._internal.utils.misc import get_prog, normalize_path
xuebingbing's avatar
xuebingbing committed
37 38 39 40
from pip._internal.utils.temp_dir import (
    global_tempdir_manager,
    tempdir_registry,
)
xuebingbing's avatar
xuebingbing committed
41 42 43 44
from pip._internal.utils.typing import MYPY_CHECK_RUNNING
from pip._internal.utils.virtualenv import running_under_virtualenv

if MYPY_CHECK_RUNNING:
xuebingbing's avatar
xuebingbing committed
45
    from typing import List, Optional, Tuple, Any
xuebingbing's avatar
xuebingbing committed
46 47
    from optparse import Values

xuebingbing's avatar
xuebingbing committed
48 49 50 51
    from pip._internal.utils.temp_dir import (
        TempDirectoryTypeRegistry as TempDirRegistry
    )

xuebingbing's avatar
xuebingbing committed
52 53 54 55 56 57 58 59 60 61 62 63 64 65
__all__ = ['Command']

logger = logging.getLogger(__name__)


class Command(CommandContextMixIn):
    usage = None  # type: str
    ignore_require_venv = False  # type: bool

    def __init__(self, name, summary, isolated=False):
        # type: (str, str, bool) -> None
        super(Command, self).__init__()
        parser_kw = {
            'usage': self.usage,
xuebingbing's avatar
xuebingbing committed
66
            'prog': '{} {}'.format(get_prog(), name),
xuebingbing's avatar
xuebingbing committed
67 68 69 70 71 72 73 74 75 76 77
            'formatter': UpdatingDefaultsHelpFormatter(),
            'add_help_option': False,
            'name': name,
            'description': self.__doc__,
            'isolated': isolated,
        }

        self.name = name
        self.summary = summary
        self.parser = ConfigOptionParser(**parser_kw)

xuebingbing's avatar
xuebingbing committed
78 79
        self.tempdir_registry = None  # type: Optional[TempDirRegistry]

xuebingbing's avatar
xuebingbing committed
80
        # Commands should add options to this option group
xuebingbing's avatar
xuebingbing committed
81
        optgroup_name = '{} Options'.format(self.name.capitalize())
xuebingbing's avatar
xuebingbing committed
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
        self.cmd_opts = optparse.OptionGroup(self.parser, optgroup_name)

        # Add the general options
        gen_opts = cmdoptions.make_option_group(
            cmdoptions.general_group,
            self.parser,
        )
        self.parser.add_option_group(gen_opts)

    def handle_pip_version_check(self, options):
        # type: (Values) -> None
        """
        This is a no-op so that commands by default do not do the pip version
        check.
        """
        # Make sure we do the pip version check if the index_group options
        # are present.
        assert not hasattr(options, 'no_index')

    def run(self, options, args):
        # type: (Values, List[Any]) -> Any
        raise NotImplementedError

    def parse_args(self, args):
        # type: (List[str]) -> Tuple[Any, Any]
        # factored out for testability
        return self.parser.parse_args(args)

    def main(self, args):
        # type: (List[str]) -> int
        try:
            with self.main_context():
                return self._main(args)
        finally:
            logging.shutdown()

    def _main(self, args):
        # type: (List[str]) -> int
xuebingbing's avatar
xuebingbing committed
120 121 122 123
        # We must initialize this before the tempdir manager, otherwise the
        # configuration would not be accessible by the time we clean up the
        # tempdir manager.
        self.tempdir_registry = self.enter_context(tempdir_registry())
xuebingbing's avatar
xuebingbing committed
124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143
        # Intentionally set as early as possible so globally-managed temporary
        # directories are available to the rest of the code.
        self.enter_context(global_tempdir_manager())

        options, args = self.parse_args(args)

        # Set verbosity so that it can be used elsewhere.
        self.verbosity = options.verbose - options.quiet

        level_number = setup_logging(
            verbosity=self.verbosity,
            no_color=options.no_color,
            user_log_file=options.log,
        )

        if (
            sys.version_info[:2] == (2, 7) and
            not options.no_python_version_warning
        ):
            message = (
xuebingbing's avatar
xuebingbing committed
144
                "pip 21.0 will drop support for Python 2.7 in January 2021. "
xuebingbing's avatar
xuebingbing committed
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
                "More details about Python 2 support in pip, can be found at "
                "https://pip.pypa.io/en/latest/development/release-process/#python-2-support"  # noqa
            )
            if platform.python_implementation() == "CPython":
                message = (
                    "Python 2.7 reached the end of its life on January "
                    "1st, 2020. Please upgrade your Python as Python 2.7 "
                    "is no longer maintained. "
                ) + message
            deprecated(message, replacement=None, gone_in=None)

        # TODO: Try to get these passing down from the command?
        #       without resorting to os.environ to hold these.
        #       This also affects isolated builds and it should.

        if options.no_input:
            os.environ['PIP_NO_INPUT'] = '1'

        if options.exists_action:
            os.environ['PIP_EXISTS_ACTION'] = ' '.join(options.exists_action)

        if options.require_venv and not self.ignore_require_venv:
            # If a venv is required check if it can really be found
            if not running_under_virtualenv():
                logger.critical(
                    'Could not find an activated virtualenv (required).'
                )
                sys.exit(VIRTUALENV_NOT_FOUND)

        if options.cache_dir:
            options.cache_dir = normalize_path(options.cache_dir)
            if not check_path_owner(options.cache_dir):
                logger.warning(
                    "The directory '%s' or its parent directory is not owned "
                    "or is not writable by the current user. The cache "
                    "has been disabled. Check the permissions and owner of "
                    "that directory. If executing pip with sudo, you may want "
                    "sudo's -H flag.",
                    options.cache_dir,
                )
                options.cache_dir = None

        try:
            status = self.run(options, args)
            # FIXME: all commands should return an exit status
            # and when it is done, isinstance is not needed anymore
            if isinstance(status, int):
                return status
        except PreviousBuildDirError as exc:
            logger.critical(str(exc))
            logger.debug('Exception information:', exc_info=True)

            return PREVIOUS_BUILD_DIR_ERROR
        except (InstallationError, UninstallationError, BadCommand) as exc:
            logger.critical(str(exc))
            logger.debug('Exception information:', exc_info=True)

            return ERROR
        except CommandError as exc:
            logger.critical('%s', exc)
            logger.debug('Exception information:', exc_info=True)

            return ERROR
        except BrokenStdoutLoggingError:
            # Bypass our logger and write any remaining messages to stderr
            # because stdout no longer works.
            print('ERROR: Pipe to stdout was broken', file=sys.stderr)
            if level_number <= logging.DEBUG:
                traceback.print_exc(file=sys.stderr)

            return ERROR
        except KeyboardInterrupt:
            logger.critical('Operation cancelled by user')
            logger.debug('Exception information:', exc_info=True)

            return ERROR
        except BaseException:
            logger.critical('Exception:', exc_info=True)

            return UNKNOWN_ERROR
        finally:
            self.handle_pip_version_check(options)

        return SUCCESS