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

1"""Structured JSON logging for basanos. 

2 

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

4:class:`logging.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 _to_serialisable(value: Any) -> Any: 

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

63 

64 Non-finite :class:`float` values (``nan``, ``inf``, ``-inf``) are 

65 converted to their :func:`str` representation so that the resulting JSON 

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

67 :class:`dict`, :class:`list`, and :class:`tuple` containers are traversed 

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

69 ``default=str`` fallback in :func:`json.dumps`. 

70 

71 Args: 

72 value: The value to coerce. 

73 

74 Returns: 

75 A JSON-serialisable representation of *value*. 

76 """ 

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

78 return str(value) 

79 if isinstance(value, dict): 

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

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

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

83 return value 

84 

85 

86class JSONFormatter(logging.Formatter): 

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

88 

89 Applications can attach this formatter to any :class:`logging.Handler` to 

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

91 

92 The JSON payload always contains: 

93 

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

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

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

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

98 

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

100 argument to :meth:`logging.Logger.warning` (or equivalent) are merged 

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

102 structured context is ``"context"`` (a plain :class:`dict`), but any 

103 JSON-serialisable extra key is accepted. 

104 

105 Non-finite :class:`float` values (``nan``, ``inf``) and other 

106 non-serialisable types are converted to their :func:`str` representation 

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

108 :data:`math.nan`, :class:`numpy.float64`, :class:`datetime.date`). 

109 

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

111 ``Infinity`` tokens). 

112 

113 Example:: 

114 

115 handler = logging.StreamHandler() 

116 handler.setFormatter(JSONFormatter()) 

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

118 

119 Args: 

120 datefmt: Optional :func:`time.strftime` format string for the 

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

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

123 """ 

124 

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

126 

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

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

129 

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

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

132 

133 Args: 

134 record: The :class:`~logging.LogRecord` to format. 

135 

136 Returns: 

137 A single-line JSON string. 

138 """ 

139 payload: dict[str, Any] = { 

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

141 "level": record.levelname, 

142 "logger": record.name, 

143 "event": record.getMessage(), 

144 } 

145 

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

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

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

149 payload[key] = _to_serialisable(value) 

150 

151 if record.exc_info: 

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

153 

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