From 3773c7aad624588903108ee9711566c582da1fa2 Mon Sep 17 00:00:00 2001 From: Polina Date: Thu, 2 Jul 2026 18:58:31 +0000 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?venv/Lib/site-packages/pip/=5Finternal/cli=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../pip/_internal/cli/index_command.py | 195 ++++++++++ .../site-packages/pip/_internal/cli/main.py | 85 +++++ .../pip/_internal/cli/main_parser.py | 136 +++++++ .../site-packages/pip/_internal/cli/parser.py | 358 ++++++++++++++++++ .../pip/_internal/cli/progress_bars.py | 153 ++++++++ 5 files changed, 927 insertions(+) create mode 100644 venv/Lib/site-packages/pip/_internal/cli/index_command.py create mode 100644 venv/Lib/site-packages/pip/_internal/cli/main.py create mode 100644 venv/Lib/site-packages/pip/_internal/cli/main_parser.py create mode 100644 venv/Lib/site-packages/pip/_internal/cli/parser.py create mode 100644 venv/Lib/site-packages/pip/_internal/cli/progress_bars.py diff --git a/venv/Lib/site-packages/pip/_internal/cli/index_command.py b/venv/Lib/site-packages/pip/_internal/cli/index_command.py new file mode 100644 index 0000000..0f1ee1a --- /dev/null +++ b/venv/Lib/site-packages/pip/_internal/cli/index_command.py @@ -0,0 +1,195 @@ +""" +Contains command classes which may interact with an index / the network. + +Unlike its sister module, req_command, this module still uses lazy imports +so commands which don't always hit the network (e.g. list w/o --outdated or +--uptodate) don't need waste time importing PipSession and friends. +""" + +from __future__ import annotations + +import logging +import os +import sys +from functools import lru_cache +from optparse import Values +from typing import TYPE_CHECKING + +from pip._vendor import certifi + +from pip._internal.cli.base_command import Command +from pip._internal.cli.command_context import CommandContextMixIn + +if TYPE_CHECKING: + from ssl import SSLContext + + from pip._vendor.packaging.utils import NormalizedName + + from pip._internal.network.session import PipSession + +logger = logging.getLogger(__name__) + + +@lru_cache +def _create_truststore_ssl_context() -> SSLContext | None: + if sys.version_info < (3, 10): + logger.debug("Disabling truststore because Python version isn't 3.10+") + return None + + try: + import ssl + except ImportError: + logger.warning("Disabling truststore since ssl support is missing") + return None + + try: + from pip._vendor import truststore + except ImportError: + logger.warning("Disabling truststore because platform isn't supported") + return None + + ctx = truststore.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(certifi.where()) + return ctx + + +class SessionCommandMixin(CommandContextMixIn): + """ + A class mixin for command classes needing _build_session(). + """ + + def __init__(self) -> None: + super().__init__() + self._session: PipSession | None = None + + @classmethod + def _get_index_urls(cls, options: Values) -> list[str] | None: + """Return a list of index urls from user-provided options.""" + index_urls = [] + if not getattr(options, "no_index", False): + url = getattr(options, "index_url", None) + if url: + index_urls.append(url) + urls = getattr(options, "extra_index_urls", None) + if urls: + index_urls.extend(urls) + # Return None rather than an empty list + return index_urls or None + + def get_default_session(self, options: Values) -> PipSession: + """Get a default-managed session.""" + if self._session is None: + self._session = self.enter_context(self._build_session(options)) + # there's no type annotation on requests.Session, so it's + # automatically ContextManager[Any] and self._session becomes Any, + # then https://github.com/python/mypy/issues/7696 kicks in + assert self._session is not None + return self._session + + def _build_session( + self, + options: Values, + retries: int | None = None, + timeout: int | None = None, + ) -> PipSession: + from pip._internal.network.session import PipSession + + cache_dir = options.cache_dir + assert not cache_dir or os.path.isabs(cache_dir) + + if "legacy-certs" not in options.deprecated_features_enabled: + ssl_context = _create_truststore_ssl_context() + else: + ssl_context = None + + session = PipSession( + cache=os.path.join(cache_dir, "http-v2") if cache_dir else None, + retries=retries if retries is not None else options.retries, + resume_retries=options.resume_retries, + trusted_hosts=options.trusted_hosts, + index_urls=self._get_index_urls(options), + ssl_context=ssl_context, + ) + + # Handle custom ca-bundles from the user + if options.cert: + session.verify = options.cert + + # Handle SSL client certificate + if options.client_cert: + session.cert = options.client_cert + + # Handle timeouts + if options.timeout or timeout: + session.timeout = timeout if timeout is not None else options.timeout + + # Handle configured proxies + if options.proxy: + session.proxies = { + "http": options.proxy, + "https": options.proxy, + } + session.trust_env = False + session.pip_proxy = options.proxy + + # Determine if we can prompt the user for authentication or not + session.auth.prompting = not options.no_input + session.auth.keyring_provider = options.keyring_provider + + return session + + +def _pip_self_version_check(session: PipSession, options: Values) -> None: + from pip._internal.self_outdated_check import pip_self_version_check as check + + check(session, options) + + +class IndexGroupCommand(Command, SessionCommandMixin): + """ + Abstract base class for commands with the index_group options. + + This also corresponds to the commands that permit the pip version check. + """ + + def should_exclude_prerelease( + self, options: Values, package_name: NormalizedName + ) -> bool: + """ + Determine if pre-releases should be excluded for a package. + """ + # Check per-package release control settings + if options.release_control: + allow_prereleases = options.release_control.allows_prereleases(package_name) + if allow_prereleases is True: + return False # Include pre-releases + elif allow_prereleases is False: + return True # Exclude pre-releases + + # No specific setting: exclude prereleases by default + return True + + def handle_pip_version_check(self, options: Values) -> None: + """ + Do the pip version check if not disabled. + + This overrides the default behavior of not doing the check. + """ + # Make sure the index_group options are present. + assert hasattr(options, "no_index") + + if options.disable_pip_version_check or options.no_index: + return + + try: + # Otherwise, check if we're using the latest version of pip available. + session = self._build_session( + options, + retries=0, + timeout=min(5, options.timeout), + ) + with session: + _pip_self_version_check(session, options) + except Exception: + logger.warning("There was an error checking the latest version of pip.") + logger.debug("See below for error", exc_info=True) diff --git a/venv/Lib/site-packages/pip/_internal/cli/main.py b/venv/Lib/site-packages/pip/_internal/cli/main.py new file mode 100644 index 0000000..78c30c3 --- /dev/null +++ b/venv/Lib/site-packages/pip/_internal/cli/main.py @@ -0,0 +1,85 @@ +"""Primary application entrypoint.""" + +from __future__ import annotations + +import locale +import logging +import os +import sys +import warnings + +logger = logging.getLogger(__name__) + + +# Do not import and use main() directly! Using it directly is actively +# discouraged by pip's maintainers. The name, location and behavior of +# this function is subject to change, so calling it directly is not +# portable across different pip versions. + +# In addition, running pip in-process is unsupported and unsafe. This is +# elaborated in detail at +# https://pip.pypa.io/en/stable/user_guide/#using-pip-from-your-program. +# That document also provides suggestions that should work for nearly +# all users that are considering importing and using main() directly. + +# However, we know that certain users will still want to invoke pip +# in-process. If you understand and accept the implications of using pip +# in an unsupported manner, the best approach is to use runpy to avoid +# depending on the exact location of this entry point. + +# The following example shows how to use runpy to invoke pip in that +# case: +# +# sys.argv = ["pip", your, args, here] +# runpy.run_module("pip", run_name="__main__") +# +# Note that this will exit the process after running, unlike a direct +# call to main. As it is not safe to do any processing after calling +# main, this should not be an issue in practice. + + +def main(args: list[str] | None = None) -> int: + # NOTE: Lazy imports to speed up import of this module, + # which is imported from the pip console script. This doesn't + # speed up normal pip execution, but might be important in the future + # if we use ``multiprocessing`` module, + # which imports __main__ for each spawned subprocess. + from pip._internal.cli.autocompletion import autocomplete + from pip._internal.cli.main_parser import parse_command + from pip._internal.commands import create_command + from pip._internal.exceptions import PipError + from pip._internal.utils import deprecation + + if args is None: + args = sys.argv[1:] + + # Suppress the pkg_resources deprecation warning + # Note - we use a module of .*pkg_resources to cover + # the normal case (pip._vendor.pkg_resources) and the + # devendored case (a bare pkg_resources) + warnings.filterwarnings( + action="ignore", category=DeprecationWarning, module=".*pkg_resources" + ) + + # Configure our deprecation warnings to be sent through loggers + deprecation.install_warning_logger() + + autocomplete() + + try: + cmd_name, cmd_args = parse_command(args) + except PipError as exc: + sys.stderr.write(f"ERROR: {exc}") + sys.stderr.write(os.linesep) + sys.exit(1) + + # Needed for locale.getpreferredencoding(False) to work + # in pip._internal.utils.encoding.auto_decode + try: + locale.setlocale(locale.LC_ALL, "") + except locale.Error as e: + # setlocale can apparently crash if locale are uninitialized + logger.debug("Ignoring error %s when setting locale", e) + command = create_command(cmd_name, isolated=("--isolated" in cmd_args)) + + return command.main(cmd_args) diff --git a/venv/Lib/site-packages/pip/_internal/cli/main_parser.py b/venv/Lib/site-packages/pip/_internal/cli/main_parser.py new file mode 100644 index 0000000..1368524 --- /dev/null +++ b/venv/Lib/site-packages/pip/_internal/cli/main_parser.py @@ -0,0 +1,136 @@ +"""A single place for constructing and exposing the main parser""" + +from __future__ import annotations + +import os +import subprocess +import sys + +from pip._vendor.rich.markup import escape + +from pip._internal.build_env import get_runnable_pip +from pip._internal.cli import cmdoptions +from pip._internal.cli.parser import ConfigOptionParser, UpdatingDefaultsHelpFormatter +from pip._internal.commands import commands_dict, get_similar_commands +from pip._internal.exceptions import CommandError +from pip._internal.utils.misc import get_pip_version, get_prog + +__all__ = ["create_main_parser", "parse_command"] + + +def create_main_parser() -> ConfigOptionParser: + """Creates and returns the main parser for pip's CLI""" + + parser = ConfigOptionParser( + usage="\n%prog [options]", + add_help_option=False, + formatter=UpdatingDefaultsHelpFormatter(), + name="global", + prog=get_prog(), + ) + parser.disable_interspersed_args() + + parser.version = get_pip_version() + + # add the general options + gen_opts = cmdoptions.make_option_group(cmdoptions.general_group, parser) + parser.add_option_group(gen_opts) + + # so the help formatter knows + parser.main = True # type: ignore + + # create command listing for description + description = [""] + [ + f"[optparse.longargs]{name:27}[/] {escape(command_info.summary)}" + for name, command_info in commands_dict.items() + ] + parser.description = "\n".join(description) + + return parser + + +def identify_python_interpreter(python: str) -> str | None: + # If the named file exists, use it. + # If it's a directory, assume it's a virtual environment and + # look for the environment's Python executable. + if os.path.exists(python): + if os.path.isdir(python): + # bin/python for Unix, Scripts/python.exe for Windows + # Try both in case of odd cases like cygwin. + for exe in ("bin/python", "Scripts/python.exe"): + py = os.path.join(python, exe) + if os.path.exists(py): + return py + else: + return python + + # Could not find the interpreter specified + return None + + +def parse_command(args: list[str]) -> tuple[str, list[str]]: + parser = create_main_parser() + + # Note: parser calls disable_interspersed_args(), so the result of this + # call is to split the initial args into the general options before the + # subcommand and everything else. + # For example: + # args: ['--timeout=5', 'install', '--user', 'INITools'] + # general_options: ['--timeout==5'] + # args_else: ['install', '--user', 'INITools'] + general_options, args_else = parser.parse_args(args) + + # --python + if general_options.python and "_PIP_RUNNING_IN_SUBPROCESS" not in os.environ: + # Re-invoke pip using the specified Python interpreter + interpreter = identify_python_interpreter(general_options.python) + if interpreter is None: + raise CommandError( + f"Could not locate Python interpreter {general_options.python}" + ) + + pip_cmd = [ + interpreter, + get_runnable_pip(), + ] + pip_cmd.extend(args) + + # Set a flag so the child doesn't re-invoke itself, causing + # an infinite loop. + os.environ["_PIP_RUNNING_IN_SUBPROCESS"] = "1" + returncode = 0 + try: + proc = subprocess.run(pip_cmd) + returncode = proc.returncode + except (subprocess.SubprocessError, OSError) as exc: + raise CommandError(f"Failed to run pip under {interpreter}: {exc}") + sys.exit(returncode) + + # --version + if general_options.version: + sys.stdout.write(parser.version) + sys.stdout.write(os.linesep) + sys.exit() + + # pip || pip help -> print_help() + if not args_else or (args_else[0] == "help" and len(args_else) == 1): + parser.print_help() + sys.exit() + + # the subcommand name + cmd_name = args_else[0] + + if cmd_name not in commands_dict: + guess = get_similar_commands(cmd_name) + + msg = [f'unknown command "{cmd_name}"'] + if guess: + msg.append(f'maybe you meant "{guess}"') + + raise CommandError(" - ".join(msg)) + + # all the args without the subcommand + cmd_args = args[:] + cmd_args.remove(cmd_name) + + return cmd_name, cmd_args diff --git a/venv/Lib/site-packages/pip/_internal/cli/parser.py b/venv/Lib/site-packages/pip/_internal/cli/parser.py new file mode 100644 index 0000000..a498980 --- /dev/null +++ b/venv/Lib/site-packages/pip/_internal/cli/parser.py @@ -0,0 +1,358 @@ +"""Base option parser setup""" + +from __future__ import annotations + +import logging +import optparse +import os +import re +import shutil +import sys +import textwrap +from collections.abc import Generator +from contextlib import suppress +from typing import Any, NoReturn + +from pip._vendor.rich.markup import escape +from pip._vendor.rich.theme import Theme + +from pip._internal.cli.status_codes import UNKNOWN_ERROR +from pip._internal.configuration import Configuration, ConfigurationError +from pip._internal.utils.logging import PipConsole +from pip._internal.utils.misc import redact_auth_from_url, strtobool + +logger = logging.getLogger(__name__) + + +class PrettyHelpFormatter(optparse.IndentedHelpFormatter): + """A prettier/less verbose help formatter for optparse.""" + + styles = { + "optparse.shortargs": "green", + "optparse.longargs": "cyan", + "optparse.groups": "bold blue", + "optparse.metavar": "yellow", + } + highlights = { + r"\s(-{1}[\w]+[\w-]*)": "shortargs", # highlight -letter as short args + r"\s(-{2}[\w]+[\w-]*)": "longargs", # highlight --words as long args + } + + def __init__(self, *args: Any, **kwargs: Any) -> None: + # help position must be aligned with __init__.parseopts.description + kwargs["max_help_position"] = 30 + kwargs["indent_increment"] = 1 + kwargs["width"] = shutil.get_terminal_size()[0] - 2 + super().__init__(*args, **kwargs) + + def format_option_strings(self, option: optparse.Option) -> str: + """Return a comma-separated list of option strings and metavars.""" + opts = [] + + if option._short_opts: + opts.append(f"[optparse.shortargs]{option._short_opts[0]}[/]") + if option._long_opts: + opts.append(f"[optparse.longargs]{option._long_opts[0]}[/]") + if len(opts) > 1: + opts.insert(1, ", ") + + if option.takes_value(): + assert option.dest is not None + metavar = option.metavar or option.dest.lower() + opts.append(f" [optparse.metavar]<{escape(metavar.lower())}>[/]") + + return "".join(opts) + + def format_option(self, option: optparse.Option) -> str: + """Overridden method with Rich support.""" + # fmt: off + result = [] + opts = self.option_strings[option] + opt_width = self.help_position - self.current_indent - 2 + # Remove the rich style tags before calculating width during + # text wrap calculations. Also store the length removed to adjust + # the padding in the else branch. + stripped = re.sub(r"(\[[a-z.]+\])|(\[\/\])", "", opts) + style_tag_length = len(opts) - len(stripped) + if len(stripped) > opt_width: + opts = "%*s%s\n" % (self.current_indent, "", opts) # noqa: UP031 + indent_first = self.help_position + else: # start help on same line as opts + opts = "%*s%-*s " % (self.current_indent, "", # noqa: UP031 + opt_width + style_tag_length, opts) + indent_first = 0 + result.append(opts) + if option.help: + help_text = self.expand_default(option) + help_lines = textwrap.wrap(help_text, self.help_width) + result.append("%*s%s\n" % (indent_first, "", help_lines[0])) # noqa: UP031 + result.extend(["%*s%s\n" % (self.help_position, "", line) # noqa: UP031 + for line in help_lines[1:]]) + elif opts[-1] != "\n": + result.append("\n") + return "".join(result) + # fmt: on + + def format_heading(self, heading: str) -> str: + if heading == "Options": + return "" + return "[optparse.groups]" + escape(heading) + ":[/]\n" + + def format_usage(self, usage: str) -> str: + """ + Ensure there is only one newline between usage and the first heading + if there is no description. + """ + contents = self.indent_lines(textwrap.dedent(usage), " ") + msg = f"\n[optparse.groups]Usage:[/] {escape(contents)}\n" + return msg + + def format_description(self, description: str | None) -> str: + # leave full control over description to us + if description: + if hasattr(self.parser, "main"): + label = "[optparse.groups]Commands:[/]" + else: + label = "[optparse.groups]Description:[/]" + + # some doc strings have initial newlines, some don't + description = description.lstrip("\n") + # some doc strings have final newlines and spaces, some don't + description = description.rstrip() + # dedent, then reindent + description = self.indent_lines(textwrap.dedent(description), " ") + description = f"{label}\n{description}\n" + return description + else: + return "" + + def format_epilog(self, epilog: str | None) -> str: + # leave full control over epilog to us + if epilog: + return escape(epilog) + else: + return "" + + def expand_default(self, option: optparse.Option) -> str: + """Overridden HelpFormatter.expand_default() which colorizes flags.""" + help = escape(super().expand_default(option)) + for regex, style in self.highlights.items(): + help = re.sub(regex, rf"[optparse.{style}] \1[/]", help) + return help + + def indent_lines(self, text: str, indent: str) -> str: + new_lines = [indent + line for line in text.split("\n")] + return "\n".join(new_lines) + + +class UpdatingDefaultsHelpFormatter(PrettyHelpFormatter): + """Custom help formatter for use in ConfigOptionParser. + + This is updates the defaults before expanding them, allowing + them to show up correctly in the help listing. + + Also redact auth from url type options + """ + + def expand_default(self, option: optparse.Option) -> str: + default_values = None + if self.parser is not None: + assert isinstance(self.parser, ConfigOptionParser) + self.parser._update_defaults(self.parser.defaults) + assert option.dest is not None + default_values = self.parser.defaults.get(option.dest) + help_text = super().expand_default(option) + + if default_values and option.metavar == "URL": + if isinstance(default_values, str): + default_values = [default_values] + + # If its not a list, we should abort and just return the help text + if not isinstance(default_values, list): + default_values = [] + + for val in default_values: + help_text = help_text.replace(val, redact_auth_from_url(val)) + + return help_text + + +class CustomOptionParser(optparse.OptionParser): + def insert_option_group( + self, idx: int, *args: Any, **kwargs: Any + ) -> optparse.OptionGroup: + """Insert an OptionGroup at a given position.""" + group = self.add_option_group(*args, **kwargs) + + self.option_groups.pop() + self.option_groups.insert(idx, group) + + return group + + @property + def option_list_all(self) -> list[optparse.Option]: + """Get a list of all options, including those in option groups.""" + res = self.option_list[:] + for i in self.option_groups: + res.extend(i.option_list) + + return res + + +class ConfigOptionParser(CustomOptionParser): + """Custom option parser which updates its defaults by checking the + configuration files and environmental variables""" + + def __init__( + self, + *args: Any, + name: str, + isolated: bool = False, + **kwargs: Any, + ) -> None: + self.name = name + self.config = Configuration(isolated) + + assert self.name + super().__init__(*args, **kwargs) + + def check_default(self, option: optparse.Option, key: str, val: Any) -> Any: + try: + return option.check_value(key, val) + except optparse.OptionValueError as exc: + print(f"An error occurred during configuration: {exc}") + sys.exit(3) + + def _get_ordered_configuration_items( + self, + ) -> Generator[tuple[str, Any], None, None]: + # Configuration gives keys in an unordered manner. Order them. + override_order = ["global", self.name, ":env:"] + + # Pool the options into different groups + # Use a dict because we need to implement the fallthrough logic after PR 12201 + # was merged which removed the fallthrough logic for options + section_items_dict: dict[str, dict[str, Any]] = { + name: {} for name in override_order + } + + for _, value in self.config.items(): + for section_key, val in value.items(): + + section, key = section_key.split(".", 1) + if section in override_order: + section_items_dict[section][key] = val + + # Now that we a dict of items per section, convert to list of tuples + # Make sure we completely remove empty values again + section_items = { + name: [(k, v) for k, v in section_items_dict[name].items() if v] + for name in override_order + } + + # Yield each group in their override order + for section in override_order: + yield from section_items[section] + + def _update_defaults(self, defaults: dict[str, Any]) -> dict[str, Any]: + """Updates the given defaults with values from the config files and + the environ. Does a little special handling for certain types of + options (lists).""" + + # Accumulate complex default state. + self.values = optparse.Values(self.defaults) + late_eval = set() + # Then set the options with those values + for key, val in self._get_ordered_configuration_items(): + # '--' because configuration supports only long names + option = self.get_option("--" + key) + + # Ignore options not present in this parser. E.g. non-globals put + # in [global] by users that want them to apply to all applicable + # commands. + if option is None: + continue + + assert option.dest is not None + + if option.action in ("store_true", "store_false"): + try: + val = strtobool(val) + except ValueError: + self.error( + f"{val} is not a valid value for {key} option, " + "please specify a boolean value like yes/no, " + "true/false or 1/0 instead." + ) + elif option.action == "count": + with suppress(ValueError): + val = strtobool(val) + with suppress(ValueError): + val = int(val) + if not isinstance(val, int) or val < 0: + self.error( + f"{val} is not a valid value for {key} option, " + "please instead specify either a non-negative integer " + "or a boolean value like yes/no or false/true " + "which is equivalent to 1/0." + ) + elif option.action == "append": + val = val.split() + val = [self.check_default(option, key, v) for v in val] + elif option.action == "callback": + assert option.callback is not None + late_eval.add(option.dest) + opt_str = option.get_opt_string() + val = option.convert_value(opt_str, val) + # From take_action + args = option.callback_args or () + kwargs = option.callback_kwargs or {} + option.callback(option, opt_str, val, self, *args, **kwargs) + else: + val = self.check_default(option, key, val) + + defaults[option.dest] = val + + for key in late_eval: + defaults[key] = getattr(self.values, key) + self.values = None + return defaults + + def get_default_values(self) -> optparse.Values: + """Overriding to make updating the defaults after instantiation of + the option parser possible, _update_defaults() does the dirty work.""" + if not self.process_default_values: + # Old, pre-Optik 1.5 behaviour. + return optparse.Values(self.defaults) + + # Load the configuration, or error out in case of an error + try: + self.config.load() + except ConfigurationError as err: + self.exit(UNKNOWN_ERROR, str(err)) + + defaults = self._update_defaults(self.defaults.copy()) # ours + for option in self._get_all_options(): + assert option.dest is not None + default = defaults.get(option.dest) + if isinstance(default, str): + opt_str = option.get_opt_string() + defaults[option.dest] = option.check_value(opt_str, default) + return optparse.Values(defaults) + + def error(self, msg: str) -> NoReturn: + self.print_usage(sys.stderr) + self.exit(UNKNOWN_ERROR, f"{msg}\n") + + def print_help(self, file: Any = None) -> None: + # This is unfortunate but necessary since arguments may have not been + # parsed yet at this point, so detect --no-color manually. + no_color = ( + "--no-color" in sys.argv + or bool(strtobool(os.environ.get("PIP_NO_COLOR", "no") or "no")) + or "NO_COLOR" in os.environ + ) + console = PipConsole( + theme=Theme(PrettyHelpFormatter.styles), no_color=no_color, file=file + ) + console.print(self.format_help().rstrip(), highlight=False) diff --git a/venv/Lib/site-packages/pip/_internal/cli/progress_bars.py b/venv/Lib/site-packages/pip/_internal/cli/progress_bars.py new file mode 100644 index 0000000..4931c85 --- /dev/null +++ b/venv/Lib/site-packages/pip/_internal/cli/progress_bars.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import functools +import sys +from collections.abc import Generator, Iterable, Iterator +from typing import TYPE_CHECKING, Callable, Literal, TypeVar + +from pip._vendor.rich.progress import ( + BarColumn, + DownloadColumn, + FileSizeColumn, + MofNCompleteColumn, + Progress, + ProgressColumn, + SpinnerColumn, + TextColumn, + TimeElapsedColumn, + TimeRemainingColumn, + TransferSpeedColumn, +) + +from pip._internal.cli.spinners import RateLimiter +from pip._internal.utils.logging import get_console, get_indentation + +if TYPE_CHECKING: + from pip._internal.req.req_install import InstallRequirement + +T = TypeVar("T") +ProgressRenderer = Callable[[Iterable[T]], Iterator[T]] +BarType = Literal["on", "off", "raw"] + + +def _rich_download_progress_bar( + iterable: Iterable[bytes], + *, + bar_type: BarType, + size: int | None, + initial_progress: int | None = None, +) -> Generator[bytes, None, None]: + assert bar_type == "on", "This should only be used in the default mode." + + if not size: + total = float("inf") + columns: tuple[ProgressColumn, ...] = ( + TextColumn("[progress.description]{task.description}"), + SpinnerColumn("line", speed=1.5), + FileSizeColumn(), + TransferSpeedColumn(), + TimeElapsedColumn(), + ) + else: + total = size + columns = ( + TextColumn("[progress.description]{task.description}"), + BarColumn(), + DownloadColumn(), + TransferSpeedColumn(), + TextColumn("{task.fields[time_description]}"), + TimeRemainingColumn(elapsed_when_finished=True), + ) + + progress = Progress(*columns, refresh_per_second=5) + task_id = progress.add_task( + " " * (get_indentation() + 2), total=total, time_description="eta" + ) + if initial_progress is not None: + progress.update(task_id, advance=initial_progress) + with progress: + for chunk in iterable: + yield chunk + progress.update(task_id, advance=len(chunk)) + progress.update(task_id, time_description="") + + +def _rich_install_progress_bar( + iterable: Iterable[InstallRequirement], *, total: int +) -> Iterator[InstallRequirement]: + columns = ( + TextColumn("{task.fields[indent]}"), + BarColumn(), + MofNCompleteColumn(), + TextColumn("{task.description}"), + ) + console = get_console() + + bar = Progress(*columns, refresh_per_second=6, console=console, transient=True) + # Hiding the progress bar at initialization forces a refresh cycle to occur + # until the bar appears, avoiding very short flashes. + task = bar.add_task("", total=total, indent=" " * get_indentation(), visible=False) + with bar: + for req in iterable: + bar.update(task, description=rf"\[{req.name}]", visible=True) + yield req + bar.advance(task) + + +def _raw_progress_bar( + iterable: Iterable[bytes], + *, + size: int | None, + initial_progress: int | None = None, +) -> Generator[bytes, None, None]: + def write_progress(current: int, total: int) -> None: + sys.stdout.write(f"Progress {current} of {total}\n") + sys.stdout.flush() + + current = initial_progress or 0 + total = size or 0 + rate_limiter = RateLimiter(0.25) + + write_progress(current, total) + for chunk in iterable: + current += len(chunk) + if rate_limiter.ready() or current == total: + write_progress(current, total) + rate_limiter.reset() + yield chunk + + +def get_download_progress_renderer( + *, bar_type: BarType, size: int | None = None, initial_progress: int | None = None +) -> ProgressRenderer[bytes]: + """Get an object that can be used to render the download progress. + + Returns a callable, that takes an iterable to "wrap". + """ + if bar_type == "on": + return functools.partial( + _rich_download_progress_bar, + bar_type=bar_type, + size=size, + initial_progress=initial_progress, + ) + elif bar_type == "raw": + return functools.partial( + _raw_progress_bar, + size=size, + initial_progress=initial_progress, + ) + else: + return iter # no-op, when passed an iterator + + +def get_install_progress_renderer( + *, bar_type: BarType, total: int +) -> ProgressRenderer[InstallRequirement]: + """Get an object that can be used to render the install progress. + Returns a callable, that takes an iterable to "wrap". + """ + if bar_type == "on": + return functools.partial(_rich_install_progress_bar, total=total) + else: + return iter