Coverage for src/basanos/_logging.py: 100%

1"""Structured JSON logging for basanos. 

2 

3Provides a `JSONFormatter` that applications can attach to any 

4`Handler` to receive log records as JSON objects with a 

5consistent schema:: 

6 

7 { 

8 "timestamp": "2024-01-01T00:00:00", 

9 "level": "WARNING", 

10 "logger": "basanos.math.optimizer", 

11 "event": "<formatted log message>", 

12 "context": { ... } # present when extra={"context": {...}} is used 

13 } 

14 

15Usage example:: 

16 

17 import logging 

18 from basanos import JSONFormatter 

19 

20 handler = logging.StreamHandler() 

21 handler.setFormatter(JSONFormatter()) 

22 logging.getLogger("basanos").addHandler(handler) 

23 logging.getLogger("basanos").setLevel(logging.DEBUG) 

24""" 

25 

26import json 

27import logging 

28import math 

29from typing import Any 

30 

31# Attributes that belong to logging.LogRecord itself and must not be 

32# re-emitted as extra context fields in the JSON payload. 

33_STDLIB_RECORD_ATTRS: frozenset[str] = frozenset( 

34 { 

35 "args", 

36 "created", 

37 "exc_info", 

38 "exc_text", 

39 "filename", 

40 "funcName", 

41 "levelname", 

42 "levelno", 

43 "lineno", 

44 "message", 

45 "module", 

46 "msecs", 

47 "msg", 

48 "name", 

49 "pathname", 

50 "process", 

51 "processName", 

52 "relativeCreated", 

53 "stack_info", 

54 "taskName", 

55 "thread", 

56 "threadName", 

57 } 

58) 

59 

60 

61def _serialise_mapping(value: dict[Any, Any]) -> dict[Any, Any]: 

62 """Recursively coerce each value of a mapping into serialisable form.""" 

63 return {k: _to_serialisable(v) for k, v in value.items()} 

64 

65 

66def _serialise_sequence(value: list[Any] | tuple[Any, ...]) -> list[Any]: 

67 """Recursively coerce each element of a sequence into serialisable form.""" 

68 return [_to_serialisable(v) for v in value] 

69 

70 

71def _to_serialisable(value: Any) -> Any: 

72 """Recursively coerce *value* into a JSON-serialisable form. 

73 

74 Non-finite `float` values (``nan``, ``inf``, ``-inf``) are 

75 converted to their `str` representation so that the resulting JSON 

76 is strictly valid (RFC 8259 does not permit ``NaN`` or ``Infinity``). 

77 `dict`, `list`, and `tuple` containers are traversed 

78 recursively; all other non-serialisable types are handled by the 

79 ``default=str`` fallback in `dumps`. 

80 

81 Args: 

82 value: The value to coerce. 

83 

84 Returns: 

85 A JSON-serialisable representation of *value*. 

86 """ 

87 if isinstance(value, float) and not math.isfinite(value): 

88 return str(value) 

89 if isinstance(value, dict): 

90 return _serialise_mapping(value) 

91 if isinstance(value, (list, tuple)): 

92 return _serialise_sequence(value) 

93 return value 

94 

95 

96class JSONFormatter(logging.Formatter): 

97 """Log formatter that serialises each record as a single-line JSON object. 

98 

99 Applications can attach this formatter to any `Handler` to 

100 receive machine-readable, structured log output from the *basanos* library. 

101 

102 The JSON payload always contains: 

103 

104 * ``timestamp`` - wall-clock time of the record formatted with *datefmt*. 

105 * ``level`` - upper-case level name (e.g. ``"WARNING"``). 

106 * ``logger`` - dotted logger name (e.g. ``"basanos.math.optimizer"``). 

107 * ``event`` - the fully-formatted log message. 

108 

109 Any extra fields supplied by the caller via the ``extra=`` keyword 

110 argument to `warning` (or equivalent) are merged 

111 into the JSON object at the top level. The conventional field for 

112 structured context is ``"context"`` (a plain `dict`), but any 

113 JSON-serialisable extra key is accepted. 

114 

115 Non-finite `float` values (``nan``, ``inf``) and other 

116 non-serialisable types are converted to their `str` representation 

117 automatically, so the formatter never raises on unexpected types (e.g. 

118 `nan`, `float64`, `date`). 

119 

120 The produced JSON is strictly RFC 8259-compliant (no bare ``NaN`` or 

121 ``Infinity`` tokens). 

122 

123 Example:: 

124 

125 handler = logging.StreamHandler() 

126 handler.setFormatter(JSONFormatter()) 

127 logging.getLogger("basanos").addHandler(handler) 

128 

129 Args: 

130 datefmt: Optional `strftime` format string for the 

131 ``timestamp`` field. Defaults to ISO-8601 

132 (``"%Y-%m-%dT%H:%M:%S"``). 

133 """ 

134 

135 _ISO_FMT = "%Y-%m-%dT%H:%M:%S" 

136 

137 def __init__(self, datefmt: str | None = None) -> None: 

138 super().__init__(datefmt=datefmt or self._ISO_FMT) 

139 

140 def format(self, record: logging.LogRecord) -> str: 

141 """Return the log record serialised as a JSON string. 

142 

143 Args: 

144 record: The `LogRecord` to format. 

145 

146 Returns: 

147 A single-line JSON string. 

148 """ 

149 payload: dict[str, Any] = { 

150 "timestamp": self.formatTime(record, self.datefmt), 

151 "level": record.levelname, 

152 "logger": record.name, 

153 "event": record.getMessage(), 

154 } 

155 

156 # Merge any extra fields supplied by the caller (e.g. "context"). 

157 for key, value in record.__dict__.items(): 

158 if key not in _STDLIB_RECORD_ATTRS and not key.startswith("_"): 

159 payload[key] = _to_serialisable(value) 

160 

161 if record.exc_info: 

162 payload["exc_info"] = self.formatException(record.exc_info) 

163 

164 return json.dumps(payload, default=str)