Coverage for /usr/local/lib/python3.12/site-packages/prefect/logging/formatters.py: 66%
51 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 10:48 +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__() 1ab
50 if fmt not in ["pretty", "default"]: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true1ab
51 raise ValueError("Format must be either 'pretty' or 'default'.")
53 self.serializer: JSONSerializer = JSONSerializer( 1ab
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] = {} 1ab
97 style_kwargs: dict[str, Any] = {} 1ab
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 true1ab
101 init_kwargs["defaults"] = defaults 1ab
102 style_kwargs["defaults"] = defaults 1ab
104 init_kwargs["validate"] = validate 1ab
106 super().__init__(format, datefmt, style, **init_kwargs) 1ab
108 self.flow_run_fmt = flow_run_fmt 1ab
109 self.task_run_fmt = task_run_fmt 1ab
111 # Retrieve the style class from the base class to avoid importing private
112 # `_STYLES` mapping
113 style_class = type(self._style) 1ab
115 self._flow_run_style = ( 1ab
116 style_class(flow_run_fmt, **style_kwargs) if flow_run_fmt else self._style
117 )
118 self._task_run_style = ( 1ab
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 true1ab
122 self._flow_run_style.validate() 1ab
123 self._task_run_style.validate() 1ab
125 def formatMessage(self, record: logging.LogRecord) -> str: 1a
126 if record.name == "prefect.flow_runs": 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true1cdefgb
127 style = self._flow_run_style
128 elif record.name == "prefect.task_runs": 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true1cdefgb
129 style = self._task_run_style
130 else:
131 style = self._style 1cdefgb
133 return style.format(record) 1cdefgb