Add machine readable download progress option

docs/html/user_guide.rst

@@ -855,6 +855,26 @@ We are using `freeze`_ here which outputs installed packages in requirements for
 
   reqs = subprocess.check_output([sys.executable, '-m', 'pip', 'freeze'])
 
+Since pip's progress bar gets hidden when running in a subprocess, you can use
+the ``--progress-bar=json`` option for easily parsable progress information::
+
+  subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'my_package', '--progress-bar=json'])
+
+Which will give the following output after it processes each download chunk:
+
+``Progress: {"current": ######, "total": ######}``
+
+Here, ``Progress:`` indicates it is download progress. The rest of the message is JSON
+with the ``current`` number of bytes downloaded and ``total`` .whl size as key/value pairs.
+Note: ``total`` is optional and may be null.
+
+This can be used to build your own progress bar, or report progress in other ways.
+This feature cannot be used unless pip is invoked in a subprocess.
+
+NOTE: Relying on the exact form of pip's output is unsupported, and so should not be used in
+production applications unless you are willing to adapt when pip's output changes.
+
+
 If you don't want to use pip's command line functionality, but are rather
 trying to implement code that works with Python packages, their metadata, or
 PyPI, then you should consider other, supported, packages that offer this type

news/11508.feature.rst

@@ -0,0 +1 @@
+Add a new progress_bar type that allows machine-readable (json) download progress

src/pip/_internal/cli/cmdoptions.py

@@ -226,9 +226,12 @@ class PipOption(Option):
     "--progress-bar",
     dest="progress_bar",
     type="choice",
-    choices=["on", "off"],
+    choices=["on", "off", "json"],
     default="on",
-    help="Specify whether the progress bar should be used [on, off] (default: on)",
+    help=(
+        "Specify whether the progress bar should be used"
+        " [on, off, json] (default: on)"
+    ),
 )
 
 log: Callable[..., Option] = partial(

src/pip/_internal/cli/progress_bars.py

@@ -1,4 +1,7 @@
 import functools
+import json
+import logging
+import sys
 from typing import Callable, Generator, Iterable, Iterator, Optional, Tuple
 
 from pip._vendor.rich.progress import (
@@ -14,8 +17,11 @@
     TransferSpeedColumn,
 )
 
+from pip._internal.cli.spinners import RateLimiter
 from pip._internal.utils.logging import get_indentation
 
+logger = logging.getLogger(__name__)
+
 DownloadProgressRenderer = Callable[[Iterable[bytes]], Iterator[bytes]]
 
 
@@ -55,6 +61,37 @@ def _rich_progress_bar(
             progress.update(task_id, advance=len(chunk))
 
 
+class _MachineReadableProgress:
+    def __init__(
+        self,
+        iterable: Iterable[bytes],
+        size: Optional[int],
+        # Copying the default from spinners.py
+        min_update_interval_seconds: float = 0.125,
+    ) -> None:
+        self._iterable = iter(iterable)
+        self._size = size
+        self._progress = 0
+        self._rate_limiter = RateLimiter(min_update_interval_seconds)
+
+    def __iter__(self) -> Iterator[bytes]:
+        return self
+
+    def __next__(self) -> bytes:
+        chunk = next(self._iterable)
+        self._progress += len(chunk)
+        progress_info = {
+            "current": self._progress,
+            "total": self._size,
+        }
+        if not self._rate_limiter.ready():
+            return chunk
+        sys.stdout.write(f"Progress: {json.dumps(progress_info)}\n")
+        sys.stdout.flush()
+        self._rate_limiter.reset()
+        return chunk
+
+
 def get_download_progress_renderer(
     *, bar_type: str, size: Optional[int] = None
 ) -> DownloadProgressRenderer:
@@ -64,5 +101,16 @@ def get_download_progress_renderer(
     """
     if bar_type == "on":
         return functools.partial(_rich_progress_bar, bar_type=bar_type, size=size)
+    elif bar_type == "json":
+        # We don't want regular users to use this progress_bar type
+        # so only use if not a TTY
+        assert (
+            not sys.stdout.isatty()
+        ), 'The "json" progress_bar type should only be used inside subprocesses.'
+        # Mimic log level
+        if logger.getEffectiveLevel() <= logging.INFO:
+            return functools.partial(_MachineReadableProgress, size=size)
+        else:
+            return iter
     else:
         return iter  # no-op, when passed an iterator