| Viewing file:  base_command.py (9.08 KB)      -rw-r--r-- Select action/file-type:
 
  (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) | 
 
"""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 typing import List, Optional, Tuple, Any
 from optparse import Values
 
 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)
 
 |