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,42 +56,6 @@ 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
@@ -109,18 +68,15 @@ loggers:
 
     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}"