Coverage for /usr/local/lib/python3.12/site-packages/prefect/logging/formatters.py: 57%
51 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 13:38 +0000
1from __future__ import annotations 1a
3import logging 1a
4import sys 1a
5import traceback 1a
6from types import TracebackType 1a
7from typing import Any, Literal, Optional, Tuple, Type, Union 1a
9import orjson 1a
11from prefect.serializers import JSONSerializer 1a
13ExceptionInfoType = Union[ 1a
14 Tuple[Type[BaseException], BaseException, Optional[TracebackType]],
15 Tuple[None, None, None],
16]
19def format_exception_info(exc_info: ExceptionInfoType) -> dict[str, Any]: 1a
20 # if sys.exc_info() returned a (None, None, None) tuple,
21 # then there's nothing to format
22 if exc_info[0] is None:
23 return {}
25 (exception_type, exception_obj, exception_traceback) = exc_info
26 return {
27 "type": exception_type.__name__,
28 "message": str(exception_obj),
29 "traceback": (
30 "".join(traceback.format_tb(exception_traceback))
31 if exception_traceback
32 else None
33 ),
34 }
37class JsonFormatter(logging.Formatter): 1a
38 """
39 Formats log records as a JSON string.
41 The format may be specified as "pretty" to format the JSON with indents and
42 newlines.
43 """
45 def __init__( 1a
46 self, fmt: Literal["pretty", "default"], dmft: str, style: str
47 ) -> None: # noqa
48 super().__init__() 1a
50 if fmt not in ["pretty", "default"]: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true1a
51 raise ValueError("Format must be either 'pretty' or 'default'.")
53 self.serializer: JSONSerializer = JSONSerializer( 1a
54 jsonlib="orjson",
55 dumps_kwargs={"option": orjson.OPT_INDENT_2} if fmt == "pretty" else {},
56 )
58 def format(self, record: logging.LogRecord) -> str: 1a
59 record_dict = record.__dict__.copy()
61 # GCP severity detection compatibility
62 record_dict.setdefault("severity", record.levelname)
64 # replace any exception tuples returned by `sys.exc_info()`
65 # with a JSON-serializable `dict`.
66 if record.exc_info:
67 record_dict["exc_info"] = format_exception_info(record.exc_info)
69 log_json_bytes = self.serializer.dumps(record_dict)
71 # JSONSerializer returns bytes; decode to string to conform to
72 # the `logging.Formatter.format` interface
73 return log_json_bytes.decode()
76class PrefectFormatter(logging.Formatter): 1a
77 def __init__( 1a
78 self,
79 format: str | None = None,
80 datefmt: str | None = None,
81 style: Literal["%", "{", "$"] = "%",
82 validate: bool = True,
83 *,
84 defaults: dict[str, Any] | None = None,
85 task_run_fmt: str | None = None,
86 flow_run_fmt: Optional[str] = None,
87 ) -> None:
88 """
89 Implementation of the standard Python formatter with support for multiple
90 message formats.
92 """
93 # See https://github.com/python/cpython/blob/c8c6113398ee9a7867fe9b08bc539cceb61e2aaa/Lib/logging/__init__.py#L546
94 # for implementation details
96 init_kwargs: dict[str, Any] = {} 1a
97 style_kwargs: dict[str, Any] = {} 1a
99 # defaults added in 3.10
100 if sys.version_info >= (3, 10): 100 ↛ 104line 100 didn't jump to line 104 because the condition on line 100 was always true1a
101 init_kwargs["defaults"] = defaults 1a
102 style_kwargs["defaults"] = defaults 1a
104 init_kwargs["validate"] = validate 1a
106 super().__init__(format, datefmt, style, **init_kwargs) 1a
108 self.flow_run_fmt = flow_run_fmt 1a
109 self.task_run_fmt = task_run_fmt 1a
111 # Retrieve the style class from the base class to avoid importing private
112 # `_STYLES` mapping
113 style_class = type(self._style) 1a
115 self._flow_run_style = ( 1a
116 style_class(flow_run_fmt, **style_kwargs) if flow_run_fmt else self._style
117 )
118 self._task_run_style = ( 1a
119 style_class(task_run_fmt, **style_kwargs) if task_run_fmt else self._style
120 )
121 if validate: 121 ↛ exitline 121 didn't return from function '__init__' because the condition on line 121 was always true1a
122 self._flow_run_style.validate() 1a
123 self._task_run_style.validate() 1a
125 def formatMessage(self, record: logging.LogRecord) -> str: 1a
126 if record.name == "prefect.flow_runs":
127 style = self._flow_run_style
128 elif record.name == "prefect.task_runs":
129 style = self._task_run_style
130 else:
131 style = self._style
133 return style.format(record)