Add get_console() to query global rich consoles

For an install progress bar, we'd like to emit logs while the progress
bar updates (for uninstallation messages, etc.). To avoid interwoven
logs, we need to log to the same console that the progress bar is using.

This is easiest to achieve by simply storing a global stdout and stderr
console, queried via a get_console() helper.

Author: ichard26

src/pip/_internal/utils/logging.py

@@ -8,7 +8,7 @@
 from dataclasses import dataclass
 from io import TextIOWrapper
 from logging import Filter
-from typing import Any, ClassVar, Generator, List, Optional, TextIO, Type
+from typing import Any, ClassVar, Generator, List, Optional, Type
 
 from pip._vendor.rich.console import (
     Console,
@@ -29,6 +29,8 @@
 from pip._internal.utils.misc import ensure_dir
 
 _log_state = threading.local()
+_stdout_console = None
+_stderr_console = None
 subprocess_logger = getLogger("pip.subprocessor")
 
 
@@ -144,12 +146,21 @@ def on_broken_pipe(self) -> None:
         raise BrokenPipeError() from None
 
 
+def get_console(*, stderr: bool = False) -> Console:
+    if stderr:
+        assert _stderr_console is not None, "stderr rich console is missing!"
+        return _stderr_console
+    else:
+        assert _stdout_console is not None, "stdout rich console is missing!"
+        return _stdout_console
+
+
 class RichPipStreamHandler(RichHandler):
     KEYWORDS: ClassVar[Optional[List[str]]] = []
 
-    def __init__(self, stream: Optional[TextIO], no_color: bool) -> None:
+    def __init__(self, console: Console) -> None:
         super().__init__(
-            console=PipConsole(file=stream, no_color=no_color, soft_wrap=True),
+            console=console,
             show_time=False,
             show_level=False,
             show_path=False,
@@ -266,17 +277,16 @@ def setup_logging(verbosity: int, no_color: bool, user_log_file: Optional[str])
     vendored_log_level = "WARNING" if level in ["INFO", "ERROR"] else "DEBUG"
 
     # Shorthands for clarity
-    log_streams = {
-        "stdout": "ext://sys.stdout",
-        "stderr": "ext://sys.stderr",
-    }
     handler_classes = {
         "stream": "pip._internal.utils.logging.RichPipStreamHandler",
         "file": "pip._internal.utils.logging.BetterRotatingFileHandler",
     }
     handlers = ["console", "console_errors", "console_subprocess"] + (
         ["user_log"] if include_user_log else []
     )
+    global _stdout_console, stderr_console
+    _stdout_console = PipConsole(file=sys.stdout, no_color=no_color, soft_wrap=True)
+    _stderr_console = PipConsole(file=sys.stderr, no_color=no_color, soft_wrap=True)
 
     logging.config.dictConfig(
         {
@@ -311,16 +321,14 @@ def setup_logging(verbosity: int, no_color: bool, user_log_file: Optional[str])
                 "console": {
                     "level": level,
                     "class": handler_classes["stream"],
-                    "no_color": no_color,
-                    "stream": log_streams["stdout"],
+                    "console": _stdout_console,
                     "filters": ["exclude_subprocess", "exclude_warnings"],
                     "formatter": "indent",
                 },
                 "console_errors": {
                     "level": "WARNING",
                     "class": handler_classes["stream"],
-                    "no_color": no_color,
-                    "stream": log_streams["stderr"],
+                    "console": _stderr_console,
                     "filters": ["exclude_subprocess"],
                     "formatter": "indent",
                 },
@@ -329,8 +337,7 @@ def setup_logging(verbosity: int, no_color: bool, user_log_file: Optional[str])
                 "console_subprocess": {
                     "level": level,
                     "class": handler_classes["stream"],
-                    "stream": log_streams["stderr"],
-                    "no_color": no_color,
+                    "console": _stderr_console,
                     "filters": ["restrict_to_subprocess"],
                     "formatter": "indent",
                 },

tests/unit/test_logging.py

@@ -10,6 +10,7 @@
 from pip._internal.utils.logging import (
     BrokenStdoutLoggingError,
     IndentingFormatter,
+    PipConsole,
     RichPipStreamHandler,
     indent_log,
 )
@@ -142,7 +143,8 @@ def test_broken_pipe_in_stderr_flush(self) -> None:
         record = self._make_log_record()
 
         with redirect_stderr(StringIO()) as stderr:
-            handler = RichPipStreamHandler(stream=stderr, no_color=True)
+            console = PipConsole(file=stderr, no_color=True, soft_wrap=True)
+            handler = RichPipStreamHandler(console)
             with patch("sys.stderr.flush") as mock_flush:
                 mock_flush.side_effect = BrokenPipeError()
                 # The emit() call raises no exception.
@@ -165,7 +167,8 @@ def test_broken_pipe_in_stdout_write(self) -> None:
         record = self._make_log_record()
 
         with redirect_stdout(StringIO()) as stdout:
-            handler = RichPipStreamHandler(stream=stdout, no_color=True)
+            console = PipConsole(file=stdout, no_color=True, soft_wrap=True)
+            handler = RichPipStreamHandler(console)
             with patch("sys.stdout.write") as mock_write:
                 mock_write.side_effect = BrokenPipeError()
                 with pytest.raises(BrokenStdoutLoggingError):
@@ -180,7 +183,8 @@ def test_broken_pipe_in_stdout_flush(self) -> None:
         record = self._make_log_record()
 
         with redirect_stdout(StringIO()) as stdout:
-            handler = RichPipStreamHandler(stream=stdout, no_color=True)
+            console = PipConsole(file=stdout, no_color=True, soft_wrap=True)
+            handler = RichPipStreamHandler(console)
             with patch("sys.stdout.flush") as mock_flush:
                 mock_flush.side_effect = BrokenPipeError()
                 with pytest.raises(BrokenStdoutLoggingError):