Detect interactive terminal to set colored logging + support override env variables

shared/logging/src/airflow_shared/logging/structlog.py

@@ -386,7 +386,7 @@ def configure_logging(
     stdlib_config: dict | None = None,
     extra_processors: Sequence[Processor] | None = None,
     callsite_parameters: Iterable[CallsiteParameter] | None = None,
-    colors: bool | None = None,
+    colors: bool = True,
     output: LogOutputType | None = None,
     namespace_log_levels: str | dict[str, str] | None = None,
     cache_logger_on_first_use: bool = True,
@@ -401,15 +401,17 @@ def configure_logging(
     :param log_level: The default log level to use for most logs
     :param log_format: A percent-style log format to write non JSON logs with.
     :param output: Where to write the logs too. If ``json_output`` is true this must be a binary stream
-    :param colors: Whether to use colors for non-JSON logs. If `None` is passed, then colors are used
-        if standard out is a TTY (that is, an interactive session).
+    :param colors: Whether to use colors for non-JSON logs. This only works if standard out is a TTY (that is,
+        an interactive session), unless overridden by environment variables described below.
+        Please note that disabling colors also disables all styling, including bold and italics.
+        The following environment variables control color behavior (set to any non-empty value to activate):
+
+        * ``NO_COLOR`` - Disables colors completely. This takes precedence over all other settings,
+        including ``FORCE_COLOR``.
 
-        It's possible to override this behavior by setting two standard environment variables to any value
-        except an empty string:
+        * ``FORCE_COLOR`` - Forces colors to be enabled, even when output is not going to a TTY. This only
+        takes effect if ``NO_COLOR`` is not set.
 
-        * ``FORCE_COLOR`` activates colors, regardless of where output is going.
-        * ``NO_COLOR`` disables colors, regardless of where the output is going and regardless the value of
-          ``FORCE_COLOR``. Please note that ``NO_COLOR`` disables all styling, including bold and italics.
     :param callsite_parameters: A list parameters about the callsite (line number, function name etc) to
         include in the logs.
 
@@ -427,11 +429,12 @@ def configure_logging(
     if "fatal" not in NAME_TO_LEVEL:
         NAME_TO_LEVEL["fatal"] = NAME_TO_LEVEL["critical"]
 
-    if colors is None:
-        colors = os.environ.get("NO_COLOR", "") == "" and (
-            os.environ.get("FORCE_COLOR", "") != ""
-            or (sys.stdout is not None and hasattr(sys.stdout, "isatty") and sys.stdout.isatty())
-        )
+    def is_atty():
+        return sys.stdout is not None and hasattr(sys.stdout, "isatty") and sys.stdout.isatty()
+
+    colors = os.environ.get("NO_COLOR", "") == "" and (
+        os.environ.get("FORCE_COLOR", "") != "" or (colors and is_atty())
+    )
 
     stdlib_config = stdlib_config or {}
     extra_processors = extra_processors or ()

shared/logging/tests/logging/test_structlog.py

@@ -25,12 +25,14 @@
 import sys
 import textwrap
 from datetime import datetime, timezone
+from unittest import mock
 
 import pytest
 import structlog
 from structlog.dev import BLUE, BRIGHT, CYAN, DIM, GREEN, MAGENTA, RESET_ALL as RESET
 from structlog.processors import CallsiteParameter
 
+from airflow_shared.logging import structlog as structlog_module
 from airflow_shared.logging.structlog import configure_logging
 
 # We don't want to use the caplog fixture in this test, as the main purpose of this file is to capture the
@@ -56,7 +58,10 @@ def configurer(**kwargs):
             else:
                 buff = io.StringIO()
 
-            configure_logging(**kwargs, output=buff)
+            with mock.patch("sys.stdout") as mock_stdout:
+                mock_stdout.isatty.return_value = True
+                configure_logging(**kwargs, output=buff)
+
             yield buff
             buff.seek(0)
         finally:
@@ -114,6 +119,48 @@ def test_colorful(structlog_config, get_logger, config_kwargs, extra_kwargs, ext
     )
 
 
+@pytest.mark.parametrize(
+    ("no_color", "force_color", "is_tty", "colors_param", "expected_colors"),
+    [
+        # NO_COLOR takes precedence over everything
+        pytest.param("1", "", True, True, False, id="no_color_set_tty_colors_true"),
+        pytest.param("1", "", True, False, False, id="no_color_set_tty_colors_false"),
+        pytest.param("1", "", False, True, False, id="no_color_set_no_tty_colors_true"),
+        pytest.param("1", "", False, False, False, id="no_color_set_no_tty_colors_false"),
+        pytest.param("1", "1", True, True, False, id="no_color_and_force_color_tty_colors_true"),
+        pytest.param("1", "1", True, False, False, id="no_color_and_force_color_tty_colors_false"),
+        pytest.param("1", "1", False, True, False, id="no_color_and_force_color_no_tty_colors_true"),
+        pytest.param("1", "1", False, False, False, id="no_color_and_force_color_no_tty_colors_false"),
+        # FORCE_COLOR takes precedence when NO_COLOR is not set
+        pytest.param("", "1", True, True, True, id="force_color_tty_colors_true"),
+        pytest.param("", "1", True, False, True, id="force_color_tty_colors_false"),
+        pytest.param("", "1", False, True, True, id="force_color_no_tty_colors_true"),
+        pytest.param("", "1", False, False, True, id="force_color_no_tty_colors_false"),
+        # When neither NO_COLOR nor FORCE_COLOR is set, check TTY and colors param
+        pytest.param("", "", True, True, True, id="tty_colors_true"),
+        pytest.param("", "", True, False, False, id="tty_colors_false"),
+        pytest.param("", "", False, True, False, id="no_tty_colors_true"),
+        pytest.param("", "", False, False, False, id="no_tty_colors_false"),
+    ],
+)
+def test_color_config(monkeypatch, no_color, force_color, is_tty, colors_param, expected_colors):
+    """Test all combinations of NO_COLOR, FORCE_COLOR, is_atty(), and colors parameter."""
+
+    monkeypatch.setenv("NO_COLOR", no_color)
+    monkeypatch.setenv("FORCE_COLOR", force_color)
+
+    with mock.patch("sys.stdout") as mock_stdout:
+        mock_stdout.isatty.return_value = is_tty
+
+        with mock.patch.object(structlog_module, "structlog_processors") as mock_processors:
+            mock_processors.return_value = ([], None, None)
+
+            structlog_module.configure_logging(colors=colors_param)
+
+            mock_processors.assert_called_once()
+            assert mock_processors.call_args.kwargs["colors"] == expected_colors
+
+
 @pytest.mark.parametrize(
     ("get_logger", "extra_kwargs", "extra_output"),
     [