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

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 typing import Any, Iterable, Mapping, Sequence 

15 

16from structlog.exceptions import DropEvent 

17 

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

19 

20 

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

22 from typing import Self 

23else: 

24 from typing_extensions import Self 

25 

26 

27class BoundLoggerBase: 

28 """ 

29 Immutable context carrier. 

30 

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

32 

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

34 - `structlog.stdlib.BoundLogger`. 

35 - `structlog.twisted.BoundLogger`, 

36 

37 See also `custom-wrappers`. 

38 """ 

39 

40 _logger: WrappedLogger 

41 """ 

42 Wrapped logger. 

43 

44 .. note:: 

45 

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

47 

48 See also `custom-wrappers`. 

49 """ 

50 

51 def __init__( 

52 self, 

53 logger: WrappedLogger, 

54 processors: Iterable[Processor], 

55 context: Context, 

56 ): 

57 self._logger = logger 

58 self._processors = processors 

59 self._context = context 

60 

61 def __repr__(self) -> str: 

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

63 

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

65 try: 

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

67 except AttributeError: 

68 return False 

69 

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

71 return not self.__eq__(other) 

72 

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

74 """ 

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

76 """ 

77 return self.__class__( 

78 self._logger, 

79 self._processors, 

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

81 ) 

82 

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

84 """ 

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

86 

87 Raises: 

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

89 """ 

90 bl = self.bind() 

91 for key in keys: 

92 del bl._context[key] 

93 

94 return bl 

95 

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

97 """ 

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

99 

100 .. versionadded:: 18.2.0 

101 """ 

102 bl = self.bind() 

103 for key in keys: 

104 bl._context.pop(key, None) 

105 

106 return bl 

107 

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

109 """ 

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

111 

112 Only necessary with dict implementations that keep global state like 

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

114 are reused. 

115 """ 

116 self._context.clear() 

117 

118 return self.bind(**new_values) 

119 

120 # Helper methods for sub-classing concrete BoundLoggers. 

121 

122 def _process_event( 

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

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

125 """ 

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

127 

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

129 process using the processor chain. 

130 

131 Args: 

132 method_name: 

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

134 

135 event: 

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

137 

138 event_kw: 

139 Additional event keywords. For example if someone calls 

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

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

142 

143 Raises: 

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

145 

146 ValueError: 

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

148 tuple, or a dict. 

149 

150 Returns: 

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

152 

153 .. note:: 

154 Despite underscore available to custom wrapper classes. 

155 

156 See also `custom-wrappers`. 

157 

158 .. versionchanged:: 14.0.0 

159 Allow final processor to return a `dict`. 

160 .. versionchanged:: 20.2.0 

161 Allow final processor to return `bytes`. 

162 .. versionchanged:: 21.2.0 

163 Allow final processor to return a `bytearray`. 

164 """ 

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

166 # EventDict. 

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

168 event_dict.update(**event_kw) 

169 

170 if event is not None: 

171 event_dict["event"] = event 

172 for proc in self._processors: 

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

174 

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

176 return (event_dict,), {} 

177 

178 if isinstance(event_dict, tuple): 

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

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

181 return event_dict 

182 

183 if isinstance(event_dict, dict): 

184 return (), event_dict 

185 

186 msg = ( 

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

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

189 ) 

190 raise ValueError(msg) 

191 

192 def _proxy_to_logger( 

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

194 ) -> Any: 

195 """ 

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

197 

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

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

200 :attr:`_logger` with the result. 

201 

202 Args: 

203 method_name: 

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

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

206 also get passed into processors. 

207 

208 event: 

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

210 

211 event_kw: 

212 Additional event keywords. For example if someone calls 

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

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

215 

216 .. note:: 

217 Despite underscore available to custom wrapper classes. 

218 

219 See also `custom-wrappers`. 

220 """ 

221 try: 

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

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

224 except DropEvent: 

225 return None 

226 

227 

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

229 """ 

230 Return *bound_logger*'s context. 

231 

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

233 configuration. 

234 

235 Args: 

236 bound_logger: The bound logger whose context you want. 

237 

238 Returns: 

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

240 

241 .. versionadded:: 20.2.0 

242 """ 

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

244 return bound_logger._context