Add PrefectFormatter to reduce logging configuration duplication

src/prefect/logging/formatters.py

@@ -1,4 +1,5 @@
 import logging.handlers
+import sys
 import traceback
 from types import TracebackType
 from typing import Optional, Tuple, Type, Union
@@ -62,3 +63,67 @@ def format(self, record: logging.LogRecord) -> str:
         # JSONSerializer returns bytes; decode to string to conform to
         # the `logging.Formatter.format` interface
         return log_json_bytes.decode()
+
+
+class PrefectFormatter(logging.Formatter):
+    def __init__(
+        self,
+        format=None,
+        datefmt=None,
+        style="%",
+        validate=True,
+        *,
+        defaults=None,
+        task_run_fmt: str = None,
+        flow_run_fmt: str = None
+    ) -> None:
+        """
+        Implementation of the standard Python formatter with support for multiple
+        message formats.
+
+        """
+        # See https://github.com/python/cpython/blob/c8c6113398ee9a7867fe9b08bc539cceb61e2aaa/Lib/logging/__init__.py#L546
+        # for implementation details
+
+        init_kwargs = {}
+        style_kwargs = {}
+
+        # defaults added in 3.10
+        if sys.version_info >= (3, 10):
+            init_kwargs["defaults"] = defaults
+            style_kwargs["defaults"] = defaults
+
+        # validate added in 3.8
+        if sys.version_info >= (3, 8):
+            init_kwargs["validate"] = validate
+        else:
+            validate = False
+
+        super().__init__(format, datefmt, style, **init_kwargs)
+
+        self.flow_run_fmt = flow_run_fmt
+        self.task_run_fmt = task_run_fmt
+
+        # Retrieve the style class from the base class to avoid importing private
+        # `_STYLES` mapping
+        style_class = type(self._style)
+
+        self._flow_run_style = (
+            style_class(flow_run_fmt, **style_kwargs) if flow_run_fmt else self._style
+        )
+        self._task_run_style = (
+            style_class(task_run_fmt, **style_kwargs) if task_run_fmt else self._style
+        )
+        if validate:
+            self._flow_run_style.validate()
+            self._task_run_style.validate()
+
+    def formatMessage(self, record: logging.LogRecord):
+        if record.name == "prefect.flow_runs":
+            style = self._flow_run_style
+        elif record.name == "prefect.task_runs":
+            style = self._task_run_style
+        else:
+            style = self._style
+
+        return style.format(record)

src/prefect/logging/logging.yml

@@ -14,15 +14,10 @@ formatters:
         datefmt: "%H:%M:%S"
 
     standard:
+        (): prefect.logging.formatters.PrefectFormatter
         format: "%(asctime)s.%(msecs)03d | %(levelname)-7s | %(name)s - %(message)s"
-        datefmt: "%H:%M:%S"
-
-    flow_runs:
-        format: "%(asctime)s.%(msecs)03d | %(levelname)-7s | Flow run %(flow_run_name)r - %(message)s"
-        datefmt: "%H:%M:%S"
-
-    task_runs:
-        format: "%(asctime)s.%(msecs)03d | %(levelname)-7s | Task run %(task_run_name)r - %(message)s"
+        flow_run_fmt: "%(asctime)s.%(msecs)03d | %(levelname)-7s | Flow run %(flow_run_name)r - %(message)s"
+        task_run_fmt: "%(asctime)s.%(msecs)03d | %(levelname)-7s | Task run %(task_run_name)r - %(message)s"
         datefmt: "%H:%M:%S"
 
     json:
@@ -61,66 +56,25 @@ handlers:
             log.flow_run_name: magenta
             log.flow_name: bold magenta
 
-    console_flow_runs:
-        level: 0
-        class: prefect.logging.handlers.PrefectConsoleHandler
-        formatter: flow_runs
-        styles:
-            log.web_url: bright_blue
-            log.local_url: bright_blue
-
-            log.info_level: cyan
-            log.warning_level: yellow3
-            log.error_level: red3
-            log.critical_level: bright_red
-
-            log.completed_state: green
-            log.cancelled_state: yellow3
-            log.failed_state: red3
-            log.crashed_state: bright_red
-
-    console_task_runs:
-        level: 0
-        class: prefect.logging.handlers.PrefectConsoleHandler
-        formatter: task_runs
-        styles:
-            log.web_url: bright_blue
-            log.local_url: bright_blue
-
-            log.info_level: cyan
-            log.warning_level: yellow3
-            log.error_level: red3
-            log.critical_level: bright_red
-
-            log.completed_state: green
-            log.cancelled_state: yellow3
-            log.failed_state: red3
-            log.state_crashed: bright_red
-
     orion:
         level: 0
         class: prefect.logging.handlers.OrionHandler
 
 loggers:
     prefect:
         level: "${PREFECT_LOGGING_LEVEL}"
-        handlers: [console]
-        propagate: no
 
     prefect.extra:
         level: "${PREFECT_LOGGING_LEVEL}"
-        handlers: [orion, console]
-        propagate: no
+        handlers: [orion]
 
     prefect.flow_runs:
         level: NOTSET
-        handlers: [orion, console_flow_runs]
-        propagate: no
+        handlers: [orion]
 
     prefect.task_runs:
         level: NOTSET
-        handlers: [orion, console_task_runs]
-        propagate: no
+        handlers: [orion]
 
     prefect.orion:
         level: "${PREFECT_LOGGING_SERVER_LEVEL}"
@@ -133,13 +87,9 @@ loggers:
 
     uvicorn:
         level: "${PREFECT_LOGGING_SERVER_LEVEL}"
-        handlers: [console]
-        propagate: no
 
     fastapi:
         level: "${PREFECT_LOGGING_SERVER_LEVEL}"
-        handlers: [console]
-        propagate: no
 
 # The root logger: any logger without propagation disabled sends to here as well
 root:

tests/conftest.py

@@ -35,7 +35,6 @@
 
 import prefect
 import prefect.settings
-from prefect.logging import get_logger
 from prefect.logging.configuration import setup_logging
 from prefect.settings import (
     PREFECT_API_URL,
@@ -465,7 +464,8 @@ def caplog(caplog):
 
     for name, logger_config in config["loggers"].items():
         if not logger_config.get("propagate", True):
-            logger = get_logger(name)
-            logger.handlers.append(caplog.handler)
+            logger = logging.getLogger(name)
+            if caplog.handler not in logger.handlers:
+                logger.handlers.append(caplog.handler)
 
     yield caplog

tests/test_logging.py

@@ -153,13 +153,13 @@ def test_setup_logging_uses_env_var_overrides(tmp_path, dictConfigMock, monkeypa
     expected_config["root"]["level"] = "ROOT_LEVEL_VAL"
 
     # Test setting a value where the a key contains underscores
-    env["PREFECT_LOGGING_FORMATTERS_FLOW_RUNS_DATEFMT"] = "UNDERSCORE_KEY_VAL"
-    expected_config["formatters"]["flow_runs"]["datefmt"] = "UNDERSCORE_KEY_VAL"
+    env["PREFECT_LOGGING_FORMATTERS_STANDARD_FLOW_RUN_FMT"] = "UNDERSCORE_KEY_VAL"
+    expected_config["formatters"]["standard"]["flow_run_fmt"] = "UNDERSCORE_KEY_VAL"
 
     # Test setting a value where the key contains a period
-    env["PREFECT_LOGGING_LOGGERS_PREFECT_FLOW_RUNS_LEVEL"] = "FLOW_RUN_VAL"
+    env["PREFECT_LOGGING_LOGGERS_PREFECT_EXTRA_LEVEL"] = "VAL"
 
-    expected_config["loggers"]["prefect.flow_runs"]["level"] = "FLOW_RUN_VAL"
+    expected_config["loggers"]["prefect.extra"]["level"] = "VAL"
 
     # Test setting a value that does not exist in the yaml config and should not be
     # set in the expected_config since there is no value to override