Загрузить файлы в «venv/Lib/site-packages/pip/_internal/cli»
This commit is contained in:
3
venv/Lib/site-packages/pip/_internal/cli/__init__.py
Normal file
3
venv/Lib/site-packages/pip/_internal/cli/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""Subpackage containing all of pip's command line interface related code"""
|
||||
|
||||
# This file intentionally does not import submodules
|
||||
184
venv/Lib/site-packages/pip/_internal/cli/autocompletion.py
Normal file
184
venv/Lib/site-packages/pip/_internal/cli/autocompletion.py
Normal file
@@ -0,0 +1,184 @@
|
||||
"""Logic that powers autocompletion installed by ``pip completion``."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import optparse
|
||||
import os
|
||||
import sys
|
||||
from collections.abc import Iterable
|
||||
from itertools import chain
|
||||
from typing import Any
|
||||
|
||||
from pip._internal.cli.main_parser import create_main_parser
|
||||
from pip._internal.commands import commands_dict, create_command
|
||||
from pip._internal.metadata import get_default_environment
|
||||
|
||||
|
||||
def autocomplete() -> None:
|
||||
"""Entry Point for completion of main and subcommand options."""
|
||||
# Don't complete if user hasn't sourced bash_completion file.
|
||||
if "PIP_AUTO_COMPLETE" not in os.environ:
|
||||
return
|
||||
# Don't complete if autocompletion environment variables
|
||||
# are not present
|
||||
if not os.environ.get("COMP_WORDS") or not os.environ.get("COMP_CWORD"):
|
||||
return
|
||||
cwords = os.environ["COMP_WORDS"].split()[1:]
|
||||
cword = int(os.environ["COMP_CWORD"])
|
||||
try:
|
||||
current = cwords[cword - 1]
|
||||
except IndexError:
|
||||
current = ""
|
||||
|
||||
parser = create_main_parser()
|
||||
subcommands = list(commands_dict)
|
||||
options = []
|
||||
|
||||
# subcommand
|
||||
subcommand_name: str | None = None
|
||||
for word in cwords:
|
||||
if word in subcommands:
|
||||
subcommand_name = word
|
||||
break
|
||||
# subcommand options
|
||||
if subcommand_name is not None:
|
||||
# special case: 'help' subcommand has no options
|
||||
if subcommand_name == "help":
|
||||
sys.exit(1)
|
||||
# special case: list locally installed dists for show and uninstall
|
||||
should_list_installed = not current.startswith("-") and subcommand_name in [
|
||||
"show",
|
||||
"uninstall",
|
||||
]
|
||||
if should_list_installed:
|
||||
env = get_default_environment()
|
||||
lc = current.lower()
|
||||
installed = [
|
||||
dist.canonical_name
|
||||
for dist in env.iter_installed_distributions(local_only=True)
|
||||
if dist.canonical_name.startswith(lc)
|
||||
and dist.canonical_name not in cwords[1:]
|
||||
]
|
||||
# if there are no dists installed, fall back to option completion
|
||||
if installed:
|
||||
for dist in installed:
|
||||
print(dist)
|
||||
sys.exit(1)
|
||||
|
||||
should_list_installables = (
|
||||
not current.startswith("-") and subcommand_name == "install"
|
||||
)
|
||||
if should_list_installables:
|
||||
for path in auto_complete_paths(current, "path"):
|
||||
print(path)
|
||||
sys.exit(1)
|
||||
|
||||
subcommand = create_command(subcommand_name)
|
||||
|
||||
for opt in subcommand.parser.option_list_all:
|
||||
if opt.help != optparse.SUPPRESS_HELP:
|
||||
options += [
|
||||
(opt_str, opt.nargs) for opt_str in opt._long_opts + opt._short_opts
|
||||
]
|
||||
|
||||
# filter out previously specified options from available options
|
||||
prev_opts = [x.split("=")[0] for x in cwords[1 : cword - 1]]
|
||||
options = [(x, v) for (x, v) in options if x not in prev_opts]
|
||||
# filter options by current input
|
||||
options = [(k, v) for k, v in options if k.startswith(current)]
|
||||
# get completion type given cwords and available subcommand options
|
||||
completion_type = get_path_completion_type(
|
||||
cwords,
|
||||
cword,
|
||||
subcommand.parser.option_list_all,
|
||||
)
|
||||
# get completion files and directories if ``completion_type`` is
|
||||
# ``<file>``, ``<dir>`` or ``<path>``
|
||||
if completion_type:
|
||||
paths = auto_complete_paths(current, completion_type)
|
||||
options = [(path, 0) for path in paths]
|
||||
for option in options:
|
||||
opt_label = option[0]
|
||||
# append '=' to options which require args
|
||||
if option[1] and option[0][:2] == "--":
|
||||
opt_label += "="
|
||||
print(opt_label)
|
||||
|
||||
# Complete sub-commands (unless one is already given).
|
||||
if not any(name in cwords for name in subcommand.handler_map()):
|
||||
for handler_name in subcommand.handler_map():
|
||||
if handler_name.startswith(current):
|
||||
print(handler_name)
|
||||
else:
|
||||
# show main parser options only when necessary
|
||||
|
||||
opts = [i.option_list for i in parser.option_groups]
|
||||
opts.append(parser.option_list)
|
||||
flattened_opts = chain.from_iterable(opts)
|
||||
if current.startswith("-"):
|
||||
for opt in flattened_opts:
|
||||
if opt.help != optparse.SUPPRESS_HELP:
|
||||
subcommands += opt._long_opts + opt._short_opts
|
||||
else:
|
||||
# get completion type given cwords and all available options
|
||||
completion_type = get_path_completion_type(cwords, cword, flattened_opts)
|
||||
if completion_type:
|
||||
subcommands = list(auto_complete_paths(current, completion_type))
|
||||
|
||||
print(" ".join([x for x in subcommands if x.startswith(current)]))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def get_path_completion_type(
|
||||
cwords: list[str], cword: int, opts: Iterable[Any]
|
||||
) -> str | None:
|
||||
"""Get the type of path completion (``file``, ``dir``, ``path`` or None)
|
||||
|
||||
:param cwords: same as the environmental variable ``COMP_WORDS``
|
||||
:param cword: same as the environmental variable ``COMP_CWORD``
|
||||
:param opts: The available options to check
|
||||
:return: path completion type (``file``, ``dir``, ``path`` or None)
|
||||
"""
|
||||
if cword < 2 or not cwords[cword - 2].startswith("-"):
|
||||
return None
|
||||
for opt in opts:
|
||||
if opt.help == optparse.SUPPRESS_HELP:
|
||||
continue
|
||||
for o in str(opt).split("/"):
|
||||
if cwords[cword - 2].split("=")[0] == o:
|
||||
if not opt.metavar or any(
|
||||
x in ("path", "file", "dir") for x in opt.metavar.split("/")
|
||||
):
|
||||
return opt.metavar
|
||||
return None
|
||||
|
||||
|
||||
def auto_complete_paths(current: str, completion_type: str) -> Iterable[str]:
|
||||
"""If ``completion_type`` is ``file`` or ``path``, list all regular files
|
||||
and directories starting with ``current``; otherwise only list directories
|
||||
starting with ``current``.
|
||||
|
||||
:param current: The word to be completed
|
||||
:param completion_type: path completion type(``file``, ``path`` or ``dir``)
|
||||
:return: A generator of regular files and/or directories
|
||||
"""
|
||||
directory, filename = os.path.split(current)
|
||||
current_path = os.path.abspath(directory)
|
||||
# Don't complete paths if they can't be accessed
|
||||
if not os.access(current_path, os.R_OK):
|
||||
return
|
||||
filename = os.path.normcase(filename)
|
||||
# list all files that start with ``filename``
|
||||
file_list = (
|
||||
x for x in os.listdir(current_path) if os.path.normcase(x).startswith(filename)
|
||||
)
|
||||
for f in file_list:
|
||||
opt = os.path.join(current_path, f)
|
||||
comp_file = os.path.normcase(os.path.join(directory, f))
|
||||
# complete regular files when there is not ``<dir>`` after option
|
||||
# complete directories when there is ``<file>``, ``<path>`` or
|
||||
# ``<dir>``after option
|
||||
if completion_type != "dir" and os.path.isfile(opt):
|
||||
yield comp_file
|
||||
elif os.path.isdir(opt):
|
||||
yield os.path.join(comp_file, "")
|
||||
255
venv/Lib/site-packages/pip/_internal/cli/base_command.py
Normal file
255
venv/Lib/site-packages/pip/_internal/cli/base_command.py
Normal file
@@ -0,0 +1,255 @@
|
||||
"""Base Command class, and related routines"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import logging.config
|
||||
import optparse
|
||||
import os
|
||||
import sys
|
||||
import traceback
|
||||
from optparse import Values
|
||||
from typing import Callable
|
||||
|
||||
from pip._vendor.rich import reconfigure
|
||||
from pip._vendor.rich import traceback as rich_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,
|
||||
DiagnosticPipError,
|
||||
InstallationError,
|
||||
NetworkConnectionError,
|
||||
PreviousBuildDirError,
|
||||
)
|
||||
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 TempDirectoryTypeRegistry as TempDirRegistry
|
||||
from pip._internal.utils.temp_dir import global_tempdir_manager, tempdir_registry
|
||||
from pip._internal.utils.virtualenv import running_under_virtualenv
|
||||
|
||||
__all__ = ["Command"]
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Command(CommandContextMixIn):
|
||||
usage: str = ""
|
||||
ignore_require_venv: bool = False
|
||||
|
||||
def __init__(self, name: str, summary: str, isolated: bool = False) -> None:
|
||||
super().__init__()
|
||||
|
||||
self.name = name
|
||||
self.summary = summary
|
||||
self.parser = ConfigOptionParser(
|
||||
usage=self.usage,
|
||||
prog=f"{get_prog()} {name}",
|
||||
formatter=UpdatingDefaultsHelpFormatter(),
|
||||
add_help_option=False,
|
||||
name=name,
|
||||
description=self.__doc__,
|
||||
isolated=isolated,
|
||||
)
|
||||
|
||||
self.tempdir_registry: TempDirRegistry | None = None
|
||||
|
||||
# Commands should add options to this option group
|
||||
optgroup_name = f"{self.name.capitalize()} Options"
|
||||
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) -> None:
|
||||
pass
|
||||
|
||||
def handle_pip_version_check(self, options: 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: Values, args: list[str]) -> int:
|
||||
raise NotImplementedError
|
||||
|
||||
def _run_wrapper(self, level_number: int, options: Values, args: list[str]) -> int:
|
||||
def _inner_run() -> int:
|
||||
try:
|
||||
return self.run(options, args)
|
||||
finally:
|
||||
self.handle_pip_version_check(options)
|
||||
|
||||
if options.debug_mode:
|
||||
rich_traceback.install(show_locals=True)
|
||||
return _inner_run()
|
||||
|
||||
try:
|
||||
status = _inner_run()
|
||||
assert isinstance(status, int)
|
||||
return status
|
||||
except DiagnosticPipError as exc:
|
||||
logger.error("%s", exc, extra={"rich": True})
|
||||
logger.debug("Exception information:", exc_info=True)
|
||||
|
||||
return ERROR
|
||||
except PreviousBuildDirError as exc:
|
||||
logger.critical(str(exc))
|
||||
logger.debug("Exception information:", exc_info=True)
|
||||
|
||||
return PREVIOUS_BUILD_DIR_ERROR
|
||||
except (
|
||||
InstallationError,
|
||||
BadCommand,
|
||||
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
|
||||
|
||||
def parse_args(self, args: list[str]) -> tuple[Values, list[str]]:
|
||||
# factored out for testability
|
||||
return self.parser.parse_args(args)
|
||||
|
||||
def main(self, args: list[str]) -> int:
|
||||
try:
|
||||
with self.main_context():
|
||||
return self._main(args)
|
||||
finally:
|
||||
logging.shutdown()
|
||||
|
||||
def _main(self, args: 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
|
||||
if options.debug_mode:
|
||||
self.verbosity = 2
|
||||
|
||||
if hasattr(options, "progress_bar") and options.progress_bar == "auto":
|
||||
options.progress_bar = "on" if self.verbosity >= 0 else "off"
|
||||
|
||||
reconfigure(no_color=options.no_color)
|
||||
level_number = setup_logging(
|
||||
verbosity=self.verbosity,
|
||||
no_color=options.no_color,
|
||||
user_log_file=options.log,
|
||||
)
|
||||
|
||||
always_enabled_features = set(options.features_enabled) & set(
|
||||
cmdoptions.ALWAYS_ENABLED_FEATURES
|
||||
)
|
||||
if always_enabled_features:
|
||||
logger.warning(
|
||||
"The following features are always enabled: %s. ",
|
||||
", ".join(sorted(always_enabled_features)),
|
||||
)
|
||||
|
||||
# Make sure that the --python argument isn't specified after the
|
||||
# subcommand. We can tell, because if --python was specified,
|
||||
# we should only reach this point if we're running in the created
|
||||
# subprocess, which has the _PIP_RUNNING_IN_SUBPROCESS environment
|
||||
# variable set.
|
||||
if options.python and "_PIP_RUNNING_IN_SUBPROCESS" not in os.environ:
|
||||
logger.critical(
|
||||
"The --python option must be placed before the pip subcommand name"
|
||||
)
|
||||
sys.exit(ERROR)
|
||||
|
||||
# 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 should "
|
||||
"use sudo's -H flag.",
|
||||
options.cache_dir,
|
||||
)
|
||||
options.cache_dir = None
|
||||
|
||||
if (
|
||||
"inprocess-build-deps" in options.features_enabled
|
||||
and os.environ.get("PIP_CONSTRAINT", "")
|
||||
and "build-constraint" not in options.features_enabled
|
||||
):
|
||||
logger.warning(
|
||||
"In-process build dependencies are enabled, "
|
||||
"PIP_CONSTRAINT will have no effect for build dependencies"
|
||||
)
|
||||
options.features_enabled.append("build-constraint")
|
||||
|
||||
return self._run_wrapper(level_number, options, args)
|
||||
|
||||
def handler_map(self) -> dict[str, Callable[[Values, list[str]], None]]:
|
||||
"""
|
||||
map of names to handler actions for commands with sub-actions
|
||||
"""
|
||||
return {}
|
||||
1267
venv/Lib/site-packages/pip/_internal/cli/cmdoptions.py
Normal file
1267
venv/Lib/site-packages/pip/_internal/cli/cmdoptions.py
Normal file
File diff suppressed because it is too large
Load Diff
28
venv/Lib/site-packages/pip/_internal/cli/command_context.py
Normal file
28
venv/Lib/site-packages/pip/_internal/cli/command_context.py
Normal file
@@ -0,0 +1,28 @@
|
||||
from collections.abc import Generator
|
||||
from contextlib import AbstractContextManager, ExitStack, contextmanager
|
||||
from typing import TypeVar
|
||||
|
||||
_T = TypeVar("_T", covariant=True)
|
||||
|
||||
|
||||
class CommandContextMixIn:
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self._in_main_context = False
|
||||
self._main_context = ExitStack()
|
||||
|
||||
@contextmanager
|
||||
def main_context(self) -> Generator[None, None, None]:
|
||||
assert not self._in_main_context
|
||||
|
||||
self._in_main_context = True
|
||||
try:
|
||||
with self._main_context:
|
||||
yield
|
||||
finally:
|
||||
self._in_main_context = False
|
||||
|
||||
def enter_context(self, context_provider: AbstractContextManager[_T]) -> _T:
|
||||
assert self._in_main_context
|
||||
|
||||
return self._main_context.enter_context(context_provider)
|
||||
Reference in New Issue
Block a user