"""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, UNKNOWN_ERROR, VIRTUALENV_NOT_FOUND, ) from pip._internal.exceptions import ( BadCommand, CommandError, InstallationError, NetworkConnectionError, PreviousBuildDirError, SubProcessError, 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 from pip._internal.utils.temp_dir import global_tempdir_manager, tempdir_registry from pip._internal.utils.typing import MYPY_CHECK_RUNNING from pip._internal.utils.virtualenv import running_under_virtualenv if MYPY_CHECK_RUNNING: from optparse import Values from typing import Any, List, Optional, Tuple from pip._internal.utils.temp_dir import ( TempDirectoryTypeRegistry as TempDirRegistry, ) __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, 'prog': '{} {}'.format(get_prog(), name), 'formatter': UpdatingDefaultsHelpFormatter(), 'add_help_option': False, 'name': name, 'description': self.__doc__, 'isolated': isolated, } self.name = name self.summary = summary self.parser = ConfigOptionParser(**parser_kw) self.tempdir_registry = None # type: Optional[TempDirRegistry] # Commands should add options to this option group optgroup_name = '{} Options'.format(self.name.capitalize()) 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) self.add_options() def add_options(self): # type: () -> None pass 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]) -> int 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 # 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()) # 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 = ( "pip 21.0 will drop support for Python 2.7 in January 2021. " "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="21.0") if ( sys.version_info[:2] == (3, 5) and not options.no_python_version_warning ): message = ( "Python 3.5 reached the end of its life on September " "13th, 2020. Please upgrade your Python as Python 3.5 " "is no longer maintained. pip 21.0 will drop support " "for Python 3.5 in January 2021." ) deprecated(message, replacement=None, gone_in="21.0") # 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 if getattr(options, "build_dir", None): deprecated( reason=( "The -b/--build/--build-dir/--build-directory " "option is deprecated." ), replacement=( "use the TMPDIR/TEMP/TMP environment variable, " "possibly combined with --no-clean" ), gone_in="20.3", issue=8333, ) if 'resolver' in options.unstable_features: logger.critical( "--unstable-feature=resolver is no longer supported, and " "has been replaced with --use-feature=2020-resolver instead." ) sys.exit(ERROR) try: status = self.run(options, args) assert 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, SubProcessError, NetworkConnectionError) 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)