Загрузить файлы в «venv/Lib/site-packages/pip/_internal/cli»

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)

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)

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 <command> [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

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)

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