Coverage for dataexcept/logging_helpers.py: 98%

1"""Helper functions for logging exceptions consistently.""" 

2 

3from __future__ import annotations 

4 

5import contextlib 

6import json 

7import logging 

8import traceback 

9from typing import Any, Iterator, Mapping, Optional 

10 

11from .redaction import redact_urls_in_text 

12 

13Context = Mapping[str, Any] 

14 

15#: Sentinel: the value could not be coerced into anything JSON will take. 

16_UNCOERCIBLE = object() 

17 

18__all__ = [ 

19 "Context", 

20 "log_and_raise", 

21 "log_exception", 

22 "log_then_raise", 

23] 

24 

25 

26def _is_json_safe(value: Any) -> bool: 

27 """True if a strict JSON encoder will accept *value* as it stands. 

28 

29 ``allow_nan=False`` because ``json.dumps`` otherwise emits bare ``NaN`` and 

30 ``Infinity``, which are not valid JSON and will be rejected downstream. 

31 """ 

32 try: 

33 json.dumps(value, allow_nan=False) 

34 except (TypeError, ValueError): 

35 return False 

36 return True 

37 

38 

39def _coerced(value: Any) -> Any: 

40 """Round-trip *value* through JSON, stringifying whatever will not encode. 

41 

42 Returns ``_UNCOERCIBLE`` rather than raising: this runs while the caller is 

43 already handling a failure. 

44 """ 

45 try: 

46 return json.loads(json.dumps(value, default=str, allow_nan=False)) 

47 except Exception: 

48 return _UNCOERCIBLE 

49 

50 

51def _described(value: Any) -> str: 

52 """Describe *value* without letting it raise. 

53 

54 An object may define a ``__repr__`` that raises. Naming the type is the 

55 most that can be said without invoking anything the object controls. 

56 """ 

57 try: 

58 return repr(value) 

59 except Exception: 

60 try: 

61 return f"<unrepresentable {type(value).__name__}>" 

62 except Exception: # pragma: no cover - a type with a hostile __name__ 

63 return "<unrepresentable>" 

64 

65 

66def _normalize_context_value(value: Any) -> Any: 

67 """Return *value* in a form a strict JSON log encoder will accept. 

68 

69 Nothing here may raise. This runs while the caller is already handling a 

70 failure, and an exception escaping would replace their error with one about 

71 logging it -- so even a hostile ``__repr__`` has to be survivable. 

72 """ 

73 if _is_json_safe(value): 

74 return value 

75 

76 coerced = _coerced(value) 

77 if coerced is not _UNCOERCIBLE: 

78 return coerced 

79 

80 return _described(value) 

81 

82 

83def _build_extra(context: Context | None) -> dict[str, Any] | None: 

84 if not context: 

85 return None 

86 serialized = { 

87 key: _normalize_context_value(value) for key, value in context.items() 

88 } 

89 return {"dataexcept_context": serialized} 

90 

91 

92def _chain_mentions_a_url(exc: BaseException) -> bool: 

93 """True if *exc* or anything it chains to renders a URL. 

94 

95 A cheap pre-check: walking the chain and testing for "://" avoids 

96 formatting a traceback for every exception that is logged. 

97 """ 

98 seen: set[int] = set() 

99 current: BaseException | None = exc 

100 while current is not None and id(current) not in seen: 

101 seen.add(id(current)) 

102 try: 

103 if "://" in str(current): 

104 return True 

105 except Exception: # pragma: no cover - a __str__ that itself raises 

106 return True 

107 current = current.__cause__ or current.__context__ 

108 return False 

109 

110 

111def log_exception( 

112 exc: Exception, 

113 logger: Optional[logging.Logger] = None, 

114 level: int = logging.ERROR, 

115 context: Context | None = None, 

116) -> None: 

117 """Log *exc* at the given log *level* using *logger*. 

118 

119 If *logger* is ``None`` a module level logger is used. 

120 

121 DataExcept redacts what it renders, but a wrapped third-party exception 

122 renders itself: an HTTP client's error may quote the credential-bearing URL 

123 it was called with, and ``exc_info`` makes logging print that whole chain. 

124 When the chain contains a URL the traceback is formatted and scrubbed here; 

125 otherwise the structured ``exc_info`` path is used unchanged, so ordinary 

126 exceptions keep the shape log aggregators expect. 

127 """ 

128 if logger is None: 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

129 logger = logging.getLogger(__name__) 

130 extra = _build_extra(context) 

131 

132 if _chain_mentions_a_url(exc): 

133 formatted = "".join( 

134 traceback.format_exception(type(exc), exc, exc.__traceback__) 

135 ) 

136 keep_path = getattr(type(exc), "_keep_url_path", True) 

137 scrubbed = redact_urls_in_text(formatted, keep_path=keep_path).rstrip() 

138 logger.log(level, "%s\n%s", exc, scrubbed, extra=extra) 

139 return 

140 

141 exc_info = (type(exc), exc, exc.__traceback__) 

142 logger.log(level, "%s", exc, exc_info=exc_info, extra=extra) 

143 

144 

145@contextlib.contextmanager 

146def log_and_raise( 

147 logger: Optional[logging.Logger] = None, 

148 level: int = logging.ERROR, 

149 context: Context | None = None, 

150) -> Iterator[None]: 

151 """Context manager that logs and re-raises exceptions preserving traceback.""" 

152 try: 

153 yield 

154 except Exception as exc: 

155 log_exception(exc, logger=logger, level=level, context=context) 

156 raise 

157 

158 

159def log_then_raise( 

160 exc: Exception, 

161 logger: Optional[logging.Logger] = None, 

162 level: int = logging.ERROR, 

163 context: Context | None = None, 

164) -> None: 

165 """Log *exc* and immediately raise it. 

166 

167 This helper mirrors the pre-context-manager API for scenarios where adding a 

168 ``with`` block would be too intrusive. Prefer :func:`log_and_raise` whenever 

169 possible so tracebacks remain untouched. 

170 """ 

171 log_exception(exc, logger=logger, level=level, context=context) 

172 raise exc