feat: Added integration with the built-in logging package

news/440.feat.md

@@ -0,0 +1 @@
+Added integration with the built-in logging package.

src/cleo/logging/cleo_handler.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+import logging
+
+from logging import LogRecord
+from typing import TYPE_CHECKING
+from typing import ClassVar
+from typing import cast
+
+from cleo.exceptions import CleoUserError
+from cleo.io.outputs.output import Verbosity
+from cleo.ui.exception_trace.component import ExceptionTrace
+
+
+if TYPE_CHECKING:
+    from cleo.io.outputs.output import Output
+
+
+class CleoFilter:
+    def __init__(self, output: Output):
+        self.output = output
+
+    @property
+    def current_loglevel(self) -> int:
+        verbosity_mapping: dict[Verbosity, int] = {
+            Verbosity.QUIET: logging.CRITICAL,  # Nothing gets emitted to the output anyway
+            Verbosity.NORMAL: logging.WARNING,
+            Verbosity.VERBOSE: logging.INFO,
+            Verbosity.VERY_VERBOSE: logging.DEBUG,
+            Verbosity.DEBUG: logging.DEBUG,
+        }
+        return verbosity_mapping[self.output.verbosity]
+
+    def filter(self, record: LogRecord) -> bool:
+        return record.levelno >= self.current_loglevel
+
+
+class CleoHandler(logging.Handler):
+    """
+    A handler class which writes logging records, appropriately formatted,
+    to a Cleo output stream.
+    """
+
+    tags: ClassVar[dict[str, str]] = {
+        "CRITICAL": "<error>",
+        "ERROR": "<error>",
+        "WARNING": "<fg=yellow>",
+        "DEBUG": "<fg=dark_gray>",
+    }
+
+    def __init__(self, output: Output):
+        super().__init__()
+        self.output = output
+        self.addFilter(CleoFilter(output))
+
+    def emit(self, record: logging.LogRecord) -> None:
+        """
+        Emit a record.
+
+        If a formatter is specified, it is used to format the record.
+        The record is then written to the output with a trailing newline.  If
+        exception information is present, it is formatted using
+        traceback.print_exception and appended to the stream.  If the stream
+        has an 'encoding' attribute, it is used to determine how to do the
+        output to the stream.
+        """
+
+        try:
+            msg = self.tags.get(record.levelname, "") + self.format(record) + "</>"
+            self.output.write(msg, new_line=True)
+            if record.exc_info:
+                _type, error, traceback = record.exc_info
+                simple = not self.output.is_verbose() or isinstance(
+                    error, CleoUserError
+                )
+                error = cast(Exception, error)
+                trace = ExceptionTrace(error)
+                trace.render(self.output, simple)
+
+        except Exception:
+            self.handleError(record)

tests/fixtures/foo4_command.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import logging
+
+from typing import TYPE_CHECKING
+from typing import ClassVar
+
+from cleo.commands.command import Command
+from cleo.helpers import option
+
+
+if TYPE_CHECKING:
+    from cleo.io.inputs.option import Option
+
+
+_logger = logging.getLogger(__file__)
+
+
+def log_stuff() -> None:
+    _logger.debug("This is an debug log record")
+    _logger.info("This is an info log record")
+    _logger.warning("This is an warning log record")
+    _logger.error("This is an error log record")
+
+
+def log_exception() -> None:
+    try:
+        raise RuntimeError("This is an exception that I raised")
+    except RuntimeError as e:
+        _logger.exception(e)
+
+
+class Foo4Command(Command):
+    name = "foo4"
+
+    description = "The foo4 bar command"
+
+    aliases: ClassVar[list[str]] = ["foo4"]
+
+    options: ClassVar[list[Option]] = [option("exception")]
+
+    def handle(self) -> int:
+        if self.option("exception"):
+            log_exception()
+        else:
+            log_stuff()
+
+        return 0

tests/test_logging.py

@@ -0,0 +1,147 @@
+from __future__ import annotations
+
+import logging
+
+import pytest
+
+from cleo.application import Application
+from cleo.logging.cleo_handler import CleoHandler
+from cleo.testers.application_tester import ApplicationTester
+from tests.fixtures.foo4_command import Foo4Command
+
+
+@pytest.fixture
+def app() -> Application:
+    app = Application()
+    cmd = Foo4Command()
+    app.add(cmd)
+    app._default_command = cmd.name
+    return app
+
+
+@pytest.fixture
+def tester(app: Application) -> ApplicationTester:
+    app.catch_exceptions(False)
+    return ApplicationTester(app)
+
+
+@pytest.fixture
+def root_logger() -> logging.Logger:
+    root = logging.getLogger()
+    root.setLevel(logging.NOTSET)
+    return root
+
+
+def test_cleohandler_normal(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("")
+
+    expected = "This is an warning log record\n" "This is an error log record\n"
+
+    assert status_code == 0
+    assert tester.io.fetch_output() == expected
+
+
+def test_cleohandler_quiet(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("-q")
+
+    assert status_code == 0
+    assert tester.io.fetch_output() == ""
+
+
+def test_cleohandler_verbose(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("-v")
+
+    expected = (
+        "This is an info log record\n"
+        "This is an warning log record\n"
+        "This is an error log record\n"
+    )
+
+    assert status_code == 0
+    assert tester.io.fetch_output() == expected
+
+
+def test_cleohandler_very_verbose(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("-vv")
+
+    expected = (
+        "This is an debug log record\n"
+        "This is an info log record\n"
+        "This is an warning log record\n"
+        "This is an error log record\n"
+    )
+
+    assert status_code == 0
+    assert tester.io.fetch_output() == expected
+
+
+def test_cleohandler_exception_normal(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("--exception")
+
+    assert status_code == 0
+    lines = tester.io.fetch_output().splitlines()
+
+    assert len(lines) == 7
+    assert lines[0] == "This is an exception that I raised"
+
+
+def test_cleohandler_exception_verbose(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("-v --exception")
+
+    assert status_code == 0
+    lines = tester.io.fetch_output().splitlines()
+
+    assert len(lines) == 20
+    assert lines[0] == "This is an exception that I raised"
+
+
+def test_cleohandler_exception_very_verbose(
+    tester: ApplicationTester,
+    root_logger: logging.Logger,
+) -> None:
+    handler = CleoHandler(tester.io.output)
+    root_logger.addHandler(handler)
+
+    status_code = tester.execute("-vv --exception")
+
+    assert status_code == 0
+    lines = tester.io.fetch_output().splitlines()
+
+    assert len(lines) == 20
+    assert lines[0] == "This is an exception that I raised"