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

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""" 

7structlog's native high-performance loggers. 

8""" 

9 

10from __future__ import annotations 

11 

12import asyncio 

13import collections 

14import contextvars 

15import sys 

16 

17from typing import Any, Callable 

18 

19from ._base import BoundLoggerBase 

20from ._log_levels import ( 

21 CRITICAL, 

22 DEBUG, 

23 ERROR, 

24 INFO, 

25 LEVEL_TO_NAME, 

26 NAME_TO_LEVEL, 

27 NOTSET, 

28 WARNING, 

29) 

30from .contextvars import _ASYNC_CALLING_STACK 

31from .typing import FilteringBoundLogger 

32 

33 

34def _nop(self: Any, event: str, *args: Any, **kw: Any) -> Any: 

35 return None 

36 

37 

38async def _anop(self: Any, event: str, *args: Any, **kw: Any) -> Any: 

39 return None 

40 

41 

42def exception( 

43 self: FilteringBoundLogger, event: str, *args: Any, **kw: Any 

44) -> Any: 

45 kw.setdefault("exc_info", True) 

46 

47 return self.error(event, *args, **kw) 

48 

49 

50async def aexception( 

51 self: FilteringBoundLogger, event: str, *args: Any, **kw: Any 

52) -> Any: 

53 """ 

54 .. versionchanged:: 23.3.0 

55 Callsite parameters are now also collected under asyncio. 

56 """ 

57 # Exception info has to be extracted this early, because it is no longer 

58 # available once control is passed to the executor. 

59 if kw.get("exc_info", True) is True: 

60 kw["exc_info"] = sys.exc_info() 

61 

62 scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] 

63 ctx = contextvars.copy_context() 

64 try: 

65 runner = await asyncio.get_running_loop().run_in_executor( 

66 None, 

67 lambda: ctx.run(lambda: self.error(event, *args, **kw)), 

68 ) 

69 finally: 

70 _ASYNC_CALLING_STACK.reset(scs_token) 

71 

72 return runner 

73 

74 

75def make_filtering_bound_logger( 

76 min_level: int | str, 

77) -> type[FilteringBoundLogger]: 

78 """ 

79 Create a new `FilteringBoundLogger` that only logs *min_level* or higher. 

80 

81 The logger is optimized such that log levels below *min_level* only consist 

82 of a ``return None``. 

83 

84 All familiar log methods are present, with async variants of each that are 

85 prefixed by an ``a``. Therefore, the async version of ``log.info("hello")`` 

86 is ``await log.ainfo("hello")``. 

87 

88 Additionally it has a ``log(self, level: int, **kw: Any)`` method to mirror 

89 `logging.Logger.log` and `structlog.stdlib.BoundLogger.log`. 

90 

91 Compared to using *structlog*'s standard library integration and the 

92 `structlog.stdlib.filter_by_level` processor: 

93 

94 - It's faster because once the logger is built at program start; it's a 

95 static class. 

96 - For the same reason you can't change the log level once configured. Use 

97 the dynamic approach of `standard-library` instead, if you need this 

98 feature. 

99 - You *can* have (much) more fine-grained filtering by :ref:`writing a 

100 simple processor <finer-filtering>`. 

101 

102 Args: 

103 min_level: 

104 The log level as an integer. You can use the constants from 

105 `logging` like ``logging.INFO`` or pass the values directly. See 

106 `this table from the logging docs 

107 <https://docs.python.org/3/library/logging.html#levels>`_ for 

108 possible values. 

109 

110 If you pass a string, it must be one of: ``critical``, ``error``, 

111 ``warning``, ``info``, ``debug``, ``notset`` (upper/lower case 

112 doesn't matter). 

113 

114 .. versionadded:: 20.2.0 

115 .. versionchanged:: 21.1.0 The returned loggers are now pickleable. 

116 .. versionadded:: 20.1.0 The ``log()`` method. 

117 .. versionadded:: 22.2.0 

118 Async variants ``alog()``, ``adebug()``, ``ainfo()``, and so forth. 

119 .. versionchanged:: 25.1.0 *min_level* can now be a string. 

120 """ 

121 if isinstance(min_level, str): 

122 min_level = NAME_TO_LEVEL[min_level.lower()] 

123 

124 return LEVEL_TO_FILTERING_LOGGER[min_level] 

125 

126 

127def _maybe_interpolate(event: str, args: tuple[Any, ...]) -> str: 

128 """ 

129 Interpolate the event string with the given arguments. 

130 

131 If there's exactly one argument and it's a mapping, use it for dict-based 

132 interpolation. Otherwise, use the arguments for positional interpolation. 

133 """ 

134 if not args: 

135 return event 

136 

137 if ( 

138 len(args) == 1 

139 and isinstance(args[0], collections.abc.Mapping) 

140 and args[0] 

141 ): 

142 return event % args[0] 

143 

144 return event % args 

145 

146 

147def _make_filtering_bound_logger(min_level: int) -> type[FilteringBoundLogger]: 

148 """ 

149 Create a new `FilteringBoundLogger` that only logs *min_level* or higher. 

150 

151 The logger is optimized such that log levels below *min_level* only consist 

152 of a ``return None``. 

153 """ 

154 

155 def make_method( 

156 level: int, 

157 ) -> tuple[Callable[..., Any], Callable[..., Any]]: 

158 if level < min_level: 

159 return _nop, _anop 

160 

161 name = LEVEL_TO_NAME[level] 

162 

163 def meth(self: Any, event: str, *args: Any, **kw: Any) -> Any: 

164 return self._proxy_to_logger( 

165 name, _maybe_interpolate(event, args), **kw 

166 ) 

167 

168 async def ameth(self: Any, event: str, *args: Any, **kw: Any) -> Any: 

169 """ 

170 .. versionchanged:: 23.3.0 

171 Callsite parameters are now also collected under asyncio. 

172 """ 

173 event = _maybe_interpolate(event, args) 

174 

175 scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] 

176 ctx = contextvars.copy_context() 

177 try: 

178 await asyncio.get_running_loop().run_in_executor( 

179 None, 

180 lambda: ctx.run( 

181 lambda: self._proxy_to_logger(name, event, **kw) 

182 ), 

183 ) 

184 finally: 

185 _ASYNC_CALLING_STACK.reset(scs_token) 

186 

187 meth.__name__ = name 

188 ameth.__name__ = f"a{name}" 

189 

190 return meth, ameth 

191 

192 def log(self: Any, level: int, event: str, *args: Any, **kw: Any) -> Any: 

193 if level < min_level: 

194 return None 

195 name = LEVEL_TO_NAME[level] 

196 

197 return self._proxy_to_logger( 

198 name, _maybe_interpolate(event, args), **kw 

199 ) 

200 

201 async def alog( 

202 self: Any, level: int, event: str, *args: Any, **kw: Any 

203 ) -> Any: 

204 """ 

205 .. versionchanged:: 23.3.0 

206 Callsite parameters are now also collected under asyncio. 

207 """ 

208 if level < min_level: 

209 return None 

210 name = LEVEL_TO_NAME[level] 

211 event = _maybe_interpolate(event, args) 

212 

213 scs_token = _ASYNC_CALLING_STACK.set(sys._getframe().f_back) # type: ignore[arg-type] 

214 ctx = contextvars.copy_context() 

215 try: 

216 runner = await asyncio.get_running_loop().run_in_executor( 

217 None, 

218 lambda: ctx.run( 

219 lambda: self._proxy_to_logger(name, event, **kw) 

220 ), 

221 ) 

222 finally: 

223 _ASYNC_CALLING_STACK.reset(scs_token) 

224 return runner 

225 

226 meths: dict[str, Callable[..., Any]] = {"log": log, "alog": alog} 

227 for lvl, name in LEVEL_TO_NAME.items(): 

228 meths[name], meths[f"a{name}"] = make_method(lvl) 

229 

230 meths["exception"] = exception 

231 meths["aexception"] = aexception 

232 meths["fatal"] = meths["critical"] 

233 meths["afatal"] = meths["acritical"] 

234 meths["warn"] = meths["warning"] 

235 meths["awarn"] = meths["awarning"] 

236 meths["msg"] = meths["info"] 

237 meths["amsg"] = meths["ainfo"] 

238 

239 # Introspection 

240 meths["is_enabled_for"] = lambda self, level: level >= min_level 

241 meths["get_effective_level"] = lambda self: min_level 

242 

243 return type( 

244 f"BoundLoggerFilteringAt{LEVEL_TO_NAME.get(min_level, 'Notset').capitalize()}", 

245 (BoundLoggerBase,), 

246 meths, 

247 ) 

248 

249 

250# Pre-create all possible filters to make them pickleable. 

251BoundLoggerFilteringAtNotset = _make_filtering_bound_logger(NOTSET) 

252BoundLoggerFilteringAtDebug = _make_filtering_bound_logger(DEBUG) 

253BoundLoggerFilteringAtInfo = _make_filtering_bound_logger(INFO) 

254BoundLoggerFilteringAtWarning = _make_filtering_bound_logger(WARNING) 

255BoundLoggerFilteringAtError = _make_filtering_bound_logger(ERROR) 

256BoundLoggerFilteringAtCritical = _make_filtering_bound_logger(CRITICAL) 

257 

258LEVEL_TO_FILTERING_LOGGER = { 

259 CRITICAL: BoundLoggerFilteringAtCritical, 

260 ERROR: BoundLoggerFilteringAtError, 

261 WARNING: BoundLoggerFilteringAtWarning, 

262 INFO: BoundLoggerFilteringAtInfo, 

263 DEBUG: BoundLoggerFilteringAtDebug, 

264 NOTSET: BoundLoggerFilteringAtNotset, 

265}