Use the rich library for logging (alt)

CHANGELOG.D/1897.bugfix

@@ -0,0 +1 @@
+Fixed conflict between logging and showing progress in `neuro cp -vv`. Use the `rich` library for outputting colored logs.

neuro-cli/src/neuro_cli/log_formatter.py

@@ -1,37 +1,47 @@
 import logging
+import sys
 
-import click
+import rich
 
 
-class ConsoleWarningFormatter(logging.Formatter):
-    """A logging.Formatter which prints WARNING and ERROR messages with
-    a prefix of the log level colored appropriate for the log level.
-    """
+class ConsoleHandler(logging.Handler):
+    def __init__(self, color: bool) -> None:
+        logging.Handler.__init__(self)
+        self.console = rich.console.Console(
+            file=sys.stderr,
+            color_system="auto" if color else None,
+            highlight=False,
+            log_path=False,
+            width=2048,
+        )
 
-    def get_level_message(self, record: logging.LogRecord) -> str:
-        separator = ": "
-
-        if record.levelno == logging.WARNING:
-            return click.style(record.levelname, fg="yellow") + separator
-        if record.levelno == logging.ERROR:
-            return click.style(record.levelname, fg="red") + separator
-
-        return ""
-
-    def format(self, record: logging.LogRecord) -> str:
-        if isinstance(record.msg, bytes):
-            record.msg = record.msg.decode("utf-8")
-        message = super().format(record)
-        return "{}{}".format(self.get_level_message(record), message)
-
-
-class ConsoleHandler(logging.StreamHandler):
     def emit(self, record: logging.LogRecord) -> None:
-        if self.stream.closed:
-            return
         try:
-            msg = self.format(record)
-            click.echo(msg, err=True)
-            self.flush()
+            self.acquire()
+            try:
+                if self.console.file.closed:
+                    return
+                self.console.print(self.get_level_message(record), end="", markup=True)
+                self.console.print(self.format(record), markup=False)
+            finally:
+                self.release()
+        except RecursionError:  # pragma: no cover
+            raise
         except Exception:  # pragma: no cover
             self.handleError(record)
+
+    def setConsole(self, console: rich.console.Console) -> None:
+        if console is not self.console:
+            self.acquire()
+            try:
+                self.console = console
+            finally:
+                self.release()
+
+    def get_level_message(self, record: logging.LogRecord) -> str:
+        if record.levelno >= logging.ERROR:
+            return f"[bold red]{record.levelname}[/bold red]: "
+        elif record.levelno >= logging.WARNING:
+            return f"[bold yellow]{record.levelname}[/bold yellow]: "
+
+        return ""

neuro-cli/src/neuro_cli/main.py

@@ -7,7 +7,7 @@
 import warnings
 from pathlib import Path
 from textwrap import dedent
-from typing import Any, List, Optional, Sequence, Tuple, Type, Union, cast
+from typing import Any, List, Optional, Sequence, Tuple, Union, cast
 
 import aiohttp
 import click
@@ -44,7 +44,7 @@
     EX_SOFTWARE,
     EX_TIMEOUT,
 )
-from .log_formatter import ConsoleHandler, ConsoleWarningFormatter
+from .log_formatter import ConsoleHandler
 from .root import Root
 from .topics import topics
 from .utils import (
@@ -89,21 +89,15 @@ def setup_stdout(errors: str) -> None:
 log = logging.getLogger(__name__)
 
 
-def setup_logging(verbosity: int, color: bool) -> None:
+def setup_logging(verbosity: int, color: bool) -> ConsoleHandler:
     root_logger = logging.getLogger()
-    handler = ConsoleHandler()
+    handler = ConsoleHandler(color=color)
     root_logger.addHandler(handler)
     root_logger.setLevel(logging.DEBUG)
 
-    if color:
-        format_class: Type[logging.Formatter] = ConsoleWarningFormatter
-    else:
-        format_class = logging.Formatter
-
-    if verbosity <= 1:
-        formatter = format_class()
-    else:
-        formatter = format_class("%(name)s.%(funcName)s: %(message)s")
+    if verbosity > 1:
+        formatter = logging.Formatter("%(name)s.%(funcName)s: %(message)s")
+        handler.setFormatter(formatter)
 
     if verbosity < -1:
         loglevel = logging.CRITICAL
@@ -116,9 +110,10 @@ def setup_logging(verbosity: int, color: bool) -> None:
     else:
         loglevel = logging.DEBUG
 
-    handler.setFormatter(formatter)
     handler.setLevel(loglevel)
 
+    return handler
+
 
 LOG_ERROR = log.error
 
@@ -159,7 +154,7 @@ def make_context(
             real_color = tty
         ctx.color = real_color
         verbosity = kwargs["verbose"] - kwargs["quiet"]
-        setup_logging(verbosity=verbosity, color=real_color)
+        handler = setup_logging(verbosity=verbosity, color=real_color)
         if kwargs["hide_token"] is None:
             hide_token_bool = True
         else:
@@ -192,6 +187,7 @@ def make_context(
             skip_gmp_stats=kwargs["skip_stats"],
             show_traceback=show_traceback,
         )
+        handler.setConsole(root.err_console)
         ctx.obj = root
         ctx.call_on_close(root.close)
         return ctx

neuro-cli/tests/unit/test_alias.py

@@ -825,7 +825,7 @@ def test_external_alias_long_option_not_identifier(
         )
         capture = run_cli(["user-cmd", "--opt", "arg"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Cannot parse option --123")
+        assert capture.err.startswith("ERROR: Cannot parse option --123")
 
     def test_external_alias_short_option_not_identifier(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -838,7 +838,7 @@ def test_external_alias_short_option_not_identifier(
         )
         capture = run_cli(["user-cmd", "--opt", "arg"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Cannot parse option -1")
+        assert capture.err.startswith("ERROR: Cannot parse option -1")
 
     def test_external_alias_option_meta_not_identifier(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -855,7 +855,7 @@ def test_external_alias_option_meta_not_identifier(
         )
         capture = run_cli(["user-cmd", "--opt", "arg"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Cannot parse option --opt 123")
+        assert capture.err.startswith("ERROR: Cannot parse option --opt 123")
 
     def test_external_alias_empty_substitution(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -864,7 +864,7 @@ def test_external_alias_empty_substitution(
         user_cfg.write_text(toml.dumps({"alias": {"user-cmd": {"exec": "script {}"}}}))
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Empty substitution is not allowed")
+        assert capture.err.startswith("ERROR: Empty substitution is not allowed")
 
     def test_external_alias_uppercased_parameter(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -875,7 +875,7 @@ def test_external_alias_uppercased_parameter(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Parameter PARAM should be lowercased")
+        assert capture.err.startswith("ERROR: Parameter PARAM should be lowercased")
 
     def test_external_alias_invalid_parameter_name(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -886,7 +886,7 @@ def test_external_alias_invalid_parameter_name(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith("Parameter 123 is not a valid identifier")
+        assert capture.err.startswith("ERROR: Parameter 123 is not a valid identifier")
 
     def test_external_alias_unknown_parameter(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -906,7 +906,9 @@ def test_external_alias_unknown_parameter(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Unknown parameter param in "script {param}"')
+        assert capture.err.startswith(
+            'ERROR: Unknown parameter param in "script {param}"'
+        )
 
     def test_external_alias_overlapped_args_and_options(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -928,7 +930,7 @@ def test_external_alias_overlapped_args_and_options(
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
         assert capture.err.startswith(
-            "The following names are present in both positional "
+            "ERROR: The following names are present in both positional "
             "and optional arguments: param"
         )
 
@@ -943,7 +945,7 @@ def test_external_alias_nested_args_brackets(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Nested brackets in "[[ARG]]"')
+        assert capture.err.startswith('ERROR: Nested brackets in "[[ARG]]"')
 
     def test_external_alias_missing_open_bracket(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -956,7 +958,7 @@ def test_external_alias_missing_open_bracket(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Missing open bracket in "ARG]"')
+        assert capture.err.startswith('ERROR: Missing open bracket in "ARG]"')
 
     def test_external_alias_missing_argument_inside_brackets(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -967,7 +969,7 @@ def test_external_alias_missing_argument_inside_brackets(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Missing argument inside brackets in "[]"')
+        assert capture.err.startswith('ERROR: Missing argument inside brackets in "[]"')
 
     def test_external_alias_ellipsis_should_follow_arg(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -979,7 +981,7 @@ def test_external_alias_ellipsis_should_follow_arg(
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
         assert capture.err.startswith(
-            'Ellipsis (...) should follow an argument in "..."'
+            'ERROR: Ellipsis (...) should follow an argument in "..."'
         )
 
     def test_external_alias_ellipsis_inside_brackets(
@@ -993,7 +995,9 @@ def test_external_alias_ellipsis_inside_brackets(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Ellipsis (...) inside of brackets in "[ARG...]"')
+        assert capture.err.startswith(
+            'ERROR: Ellipsis (...) inside of brackets in "[ARG...]"'
+        )
 
     def test_external_alias_successive_ellipsis(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -1006,7 +1010,7 @@ def test_external_alias_successive_ellipsis(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Successive ellipsis (...) in "ARG......"')
+        assert capture.err.startswith('ERROR: Successive ellipsis (...) in "ARG......"')
 
     def test_external_alias_missing_close_bracket1(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -1019,7 +1023,7 @@ def test_external_alias_missing_close_bracket1(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Missing close bracket in "[ARG"')
+        assert capture.err.startswith('ERROR: Missing close bracket in "[ARG"')
 
     def test_external_alias_missing_close_bracket2(
         self, run_cli: _RunCli, nmrc_path: Path
@@ -1032,7 +1036,7 @@ def test_external_alias_missing_close_bracket2(
         )
         capture = run_cli(["user-cmd"])
         assert capture.code == 70, capture
-        assert capture.err.startswith('Missing close bracket in "[ARG1 ARG2"')
+        assert capture.err.startswith('ERROR: Missing close bracket in "[ARG1 ARG2"')
 
 
 def test_external_alias_simplified(