Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/structlog/_native.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""" 

7structlog's native high-performance loggers. 

8""" 

9 

10from __future__ import annotations 

11 

12import asyncio 

13import contextvars 

14import sys 

15 

16from typing import Any, Callable 

17 

18from ._base import BoundLoggerBase 

19from ._log_levels import ( 

20 CRITICAL, 

21 DEBUG, 

22 ERROR, 

23 INFO, 

24 LEVEL_TO_NAME, 

25 NAME_TO_LEVEL, 

26 NOTSET, 

27 WARNING, 

28) 

29from .contextvars import _ASYNC_CALLING_STACK 

30from .typing import FilteringBoundLogger 

31 

32 

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

34 return None 

35 

36 

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

38 return None 

39 

40 

41def exception( 

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

43) -> Any: 

44 kw.setdefault("exc_info", True) 

45 

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

47 

48 

49async def aexception( 

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

51) -> Any: 

52 """ 

53 .. versionchanged:: 23.3.0 

54 Callsite parameters are now also collected under asyncio. 

55 """ 

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

57 # available once control is passed to the executor. 

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

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

60 

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

62 ctx = contextvars.copy_context() 

63 try: 

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

65 None, 

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

67 ) 

68 finally: 

69 _ASYNC_CALLING_STACK.reset(scs_token) 

70 

71 return runner 

72 

73 

74def make_filtering_bound_logger( 

75 min_level: int | str, 

76) -> type[FilteringBoundLogger]: 

77 """ 

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

79 

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

81 of a ``return None``. 

82 

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

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

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

86 

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

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

89 

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

91 `structlog.stdlib.filter_by_level` processor: 

92 

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

94 static class. 

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

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

97 feature. 

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

99 simple processor <finer-filtering>`. 

100 

101 Args: 

102 min_level: 

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

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

105 `this table from the logging docs 

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

107 possible values. 

108 

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

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

111 doesn't matter). 

112 

113 .. versionadded:: 20.2.0 

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

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

116 .. versionadded:: 22.2.0 

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

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

119 """ 

120 if isinstance(min_level, str): 

121 min_level = NAME_TO_LEVEL[min_level.lower()] 

122 

123 return LEVEL_TO_FILTERING_LOGGER[min_level] 

124 

125 

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

127 """ 

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

129 

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

131 of a ``return None``. 

132 """ 

133 

134 def make_method( 

135 level: int, 

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

137 if level < min_level: 

138 return _nop, _anop 

139 

140 name = LEVEL_TO_NAME[level] 

141 

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

143 if not args: 

144 return self._proxy_to_logger(name, event, **kw) 

145 

146 return self._proxy_to_logger(name, event % args, **kw) 

147 

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

149 """ 

150 .. versionchanged:: 23.3.0 

151 Callsite parameters are now also collected under asyncio. 

152 """ 

153 if args: 

154 event = event % args 

155 

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

157 ctx = contextvars.copy_context() 

158 try: 

159 await asyncio.get_running_loop().run_in_executor( 

160 None, 

161 lambda: ctx.run( 

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

163 ), 

164 ) 

165 finally: 

166 _ASYNC_CALLING_STACK.reset(scs_token) 

167 

168 meth.__name__ = name 

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

170 

171 return meth, ameth 

172 

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

174 if level < min_level: 

175 return None 

176 name = LEVEL_TO_NAME[level] 

177 

178 if not args: 

179 return self._proxy_to_logger(name, event, **kw) 

180 

181 return self._proxy_to_logger(name, event % args, **kw) 

182 

183 async def alog( 

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

185 ) -> Any: 

186 """ 

187 .. versionchanged:: 23.3.0 

188 Callsite parameters are now also collected under asyncio. 

189 """ 

190 if level < min_level: 

191 return None 

192 name = LEVEL_TO_NAME[level] 

193 if args: 

194 event = event % args 

195 

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

197 ctx = contextvars.copy_context() 

198 try: 

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

200 None, 

201 lambda: ctx.run( 

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

203 ), 

204 ) 

205 finally: 

206 _ASYNC_CALLING_STACK.reset(scs_token) 

207 return runner 

208 

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

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

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

212 

213 meths["exception"] = exception 

214 meths["aexception"] = aexception 

215 meths["fatal"] = meths["critical"] 

216 meths["afatal"] = meths["acritical"] 

217 meths["warn"] = meths["warning"] 

218 meths["awarn"] = meths["awarning"] 

219 meths["msg"] = meths["info"] 

220 meths["amsg"] = meths["ainfo"] 

221 

222 # Introspection 

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

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

225 

226 return type( 

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

228 (BoundLoggerBase,), 

229 meths, 

230 ) 

231 

232 

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

234BoundLoggerFilteringAtNotset = _make_filtering_bound_logger(NOTSET) 

235BoundLoggerFilteringAtDebug = _make_filtering_bound_logger(DEBUG) 

236BoundLoggerFilteringAtInfo = _make_filtering_bound_logger(INFO) 

237BoundLoggerFilteringAtWarning = _make_filtering_bound_logger(WARNING) 

238BoundLoggerFilteringAtError = _make_filtering_bound_logger(ERROR) 

239BoundLoggerFilteringAtCritical = _make_filtering_bound_logger(CRITICAL) 

240 

241LEVEL_TO_FILTERING_LOGGER = { 

242 CRITICAL: BoundLoggerFilteringAtCritical, 

243 ERROR: BoundLoggerFilteringAtError, 

244 WARNING: BoundLoggerFilteringAtWarning, 

245 INFO: BoundLoggerFilteringAtInfo, 

246 DEBUG: BoundLoggerFilteringAtDebug, 

247 NOTSET: BoundLoggerFilteringAtNotset, 

248}