Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/structlog/_base.py: 57%

1# SPDX-License-Identifier: MIT OR Apache-2.0 

2# This file is dual licensed under the terms of the Apache License, Version 

3# 2.0, and the MIT License. See the LICENSE file in the root of this 

4# repository for complete details. 

5 

6""" 

7Logger wrapper and helper class. 

8""" 

9 

10from __future__ import annotations 

11 

12import sys 

13 

14from collections.abc import Iterable, Mapping, Sequence 

15from typing import Any 

16 

17from structlog.exceptions import DropEvent 

18 

19from .typing import BindableLogger, Context, Processor, WrappedLogger 

20 

21 

22if sys.version_info >= (3, 11): 

23 from typing import Self 

24else: 

25 from typing_extensions import Self 

26 

27 

28class BoundLoggerBase: 

29 """ 

30 Immutable context carrier. 

31 

32 Doesn't do any actual logging; examples for useful subclasses are: 

33 

34 - the generic `BoundLogger` that can wrap anything, 

35 - `structlog.stdlib.BoundLogger`. 

36 - `structlog.twisted.BoundLogger`, 

37 

38 See also `custom-wrappers`. 

39 """ 

40 

41 _logger: WrappedLogger 

42 """ 

43 Wrapped logger. 

44 

45 .. note:: 

46 

47 Despite underscore available **read-only** to custom wrapper classes. 

48 

49 See also `custom-wrappers`. 

50 """ 

51 

52 def __init__( 

53 self, 

54 logger: WrappedLogger, 

55 processors: Iterable[Processor], 

56 context: Context, 

57 ): 

58 self._logger = logger 

59 self._processors = processors 

60 self._context = context 

61 

62 def __repr__(self) -> str: 

63 return f"<{self.__class__.__name__}(context={self._context!r}, processors={self._processors!r})>" 

64 

65 def __eq__(self, other: object) -> bool: 

66 try: 

67 return self._context == other._context # type: ignore[attr-defined] 

68 except AttributeError: 

69 return False 

70 

71 def __ne__(self, other: object) -> bool: 

72 return not self.__eq__(other) 

73 

74 def bind(self, **new_values: Any) -> Self: 

75 """ 

76 Return a new logger with *new_values* added to the existing ones. 

77 """ 

78 return self.__class__( 

79 self._logger, 

80 self._processors, 

81 self._context.__class__(self._context, **new_values), 

82 ) 

83 

84 def unbind(self, *keys: str) -> Self: 

85 """ 

86 Return a new logger with *keys* removed from the context. 

87 

88 Raises: 

89 KeyError: If the key is not part of the context. 

90 """ 

91 bl = self.bind() 

92 for key in keys: 

93 del bl._context[key] 

94 

95 return bl 

96 

97 def try_unbind(self, *keys: str) -> Self: 

98 """ 

99 Like :meth:`unbind`, but best effort: missing keys are ignored. 

100 

101 .. versionadded:: 18.2.0 

102 """ 

103 bl = self.bind() 

104 for key in keys: 

105 bl._context.pop(key, None) 

106 

107 return bl 

108 

109 def new(self, **new_values: Any) -> Self: 

110 """ 

111 Clear context and binds *new_values* using `bind`. 

112 

113 Only necessary with dict implementations that keep global state like 

114 those wrapped by `structlog.threadlocal.wrap_dict` when threads 

115 are reused. 

116 """ 

117 self._context.clear() 

118 

119 return self.bind(**new_values) 

120 

121 # Helper methods for sub-classing concrete BoundLoggers. 

122 

123 def _process_event( 

124 self, method_name: str, event: str | None, event_kw: dict[str, Any] 

125 ) -> tuple[Sequence[Any], Mapping[str, Any]]: 

126 """ 

127 Combines creates an ``event_dict`` and runs the chain. 

128 

129 Call it to combine your *event* and *context* into an event_dict and 

130 process using the processor chain. 

131 

132 Args: 

133 method_name: 

134 The name of the logger method. Is passed into the processors. 

135 

136 event: 

137 The event -- usually the first positional argument to a logger. 

138 

139 event_kw: 

140 Additional event keywords. For example if someone calls 

141 ``log.info("foo", bar=42)``, *event* would to be ``"foo"`` and 

142 *event_kw* ``{"bar": 42}``. 

143 

144 Raises: 

145 structlog.DropEvent: if log entry should be dropped. 

146 

147 ValueError: 

148 if the final processor doesn't return a str, bytes, bytearray, 

149 tuple, or a dict. 

150 

151 Returns: 

152 `tuple` of ``(*args, **kw)`` 

153 

154 .. note:: 

155 Despite underscore available to custom wrapper classes. 

156 

157 See also `custom-wrappers`. 

158 

159 .. versionchanged:: 14.0.0 

160 Allow final processor to return a `dict`. 

161 .. versionchanged:: 20.2.0 

162 Allow final processor to return `bytes`. 

163 .. versionchanged:: 21.2.0 

164 Allow final processor to return a `bytearray`. 

165 """ 

166 # We're typing it as Any, because processors can return more than an 

167 # EventDict. 

168 event_dict: Any = self._context.copy() 

169 event_dict.update(**event_kw) 

170 

171 if event is not None: 

172 event_dict["event"] = event 

173 for proc in self._processors: 

174 event_dict = proc(self._logger, method_name, event_dict) 

175 

176 if isinstance(event_dict, (str, bytes, bytearray)): 

177 return (event_dict,), {} 

178 

179 if isinstance(event_dict, tuple): 

180 # In this case we assume that the last processor returned a tuple 

181 # of ``(args, kwargs)`` and pass it right through. 

182 return event_dict 

183 

184 if isinstance(event_dict, dict): 

185 return (), event_dict 

186 

187 msg = ( 

188 "Last processor didn't return an appropriate value. " 

189 "Valid return values are a dict, a tuple of (args, kwargs), bytes, or a str." 

190 ) 

191 raise ValueError(msg) 

192 

193 def _proxy_to_logger( 

194 self, method_name: str, event: str | None = None, **event_kw: Any 

195 ) -> Any: 

196 """ 

197 Run processor chain on event & call *method_name* on wrapped logger. 

198 

199 DRY convenience method that runs :func:`_process_event`, takes care of 

200 handling :exc:`structlog.DropEvent`, and finally calls *method_name* on 

201 :attr:`_logger` with the result. 

202 

203 Args: 

204 method_name: 

205 The name of the method that's going to get called. Technically 

206 it should be identical to the method the user called because it 

207 also get passed into processors. 

208 

209 event: 

210 The event -- usually the first positional argument to a logger. 

211 

212 event_kw: 

213 Additional event keywords. For example if someone calls 

214 ``log.info("foo", bar=42)``, *event* would to be ``"foo"`` and 

215 *event_kw* ``{"bar": 42}``. 

216 

217 .. note:: 

218 Despite underscore available to custom wrapper classes. 

219 

220 See also `custom-wrappers`. 

221 """ 

222 try: 

223 args, kw = self._process_event(method_name, event, event_kw) 

224 return getattr(self._logger, method_name)(*args, **kw) 

225 except DropEvent: 

226 return None 

227 

228 

229def get_context(bound_logger: BindableLogger) -> Context: 

230 """ 

231 Return *bound_logger*'s context. 

232 

233 The type of *bound_logger* and the type returned depend on your 

234 configuration. 

235 

236 Args: 

237 bound_logger: The bound logger whose context you want. 

238 

239 Returns: 

240 The *actual* context from *bound_logger*. It is *not* copied first. 

241 

242 .. versionadded:: 20.2.0 

243 """ 

244 # This probably will get more complicated in the future. 

245 return bound_logger._context