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

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

7Processors and tools specific to the `Twisted <https://twisted.org/>`_ 

8networking engine. 

9 

10See also :doc:`structlog's Twisted support <twisted>`. 

11""" 

12 

13from __future__ import annotations 

14 

15import json 

16import sys 

17 

18from collections.abc import Callable, Sequence 

19from typing import Any, TextIO 

20 

21from twisted.python import log 

22from twisted.python.failure import Failure 

23from twisted.python.log import ILogObserver, textFromEventDict 

24from zope.interface import implementer 

25 

26from ._base import BoundLoggerBase 

27from ._config import _BUILTIN_DEFAULT_PROCESSORS 

28from .processors import JSONRenderer as GenericJSONRenderer 

29from .typing import EventDict, WrappedLogger 

30 

31 

32class BoundLogger(BoundLoggerBase): 

33 """ 

34 Twisted-specific version of `structlog.BoundLogger`. 

35 

36 Works exactly like the generic one except that it takes advantage of 

37 knowing the logging methods in advance. 

38 

39 Use it like:: 

40 

41 configure( 

42 wrapper_class=structlog.twisted.BoundLogger, 

43 ) 

44 

45 """ 

46 

47 def msg(self, event: str | None = None, **kw: Any) -> Any: 

48 """ 

49 Process event and call ``log.msg()`` with the result. 

50 """ 

51 return self._proxy_to_logger("msg", event, **kw) 

52 

53 def err(self, event: str | None = None, **kw: Any) -> Any: 

54 """ 

55 Process event and call ``log.err()`` with the result. 

56 """ 

57 return self._proxy_to_logger("err", event, **kw) 

58 

59 

60class LoggerFactory: 

61 """ 

62 Build a Twisted logger when an *instance* is called. 

63 

64 >>> from structlog import configure 

65 >>> from structlog.twisted import LoggerFactory 

66 >>> configure(logger_factory=LoggerFactory()) 

67 """ 

68 

69 def __call__(self, *args: Any) -> WrappedLogger: 

70 """ 

71 Positional arguments are silently ignored. 

72 

73 :rvalue: A new Twisted logger. 

74 

75 .. versionchanged:: 0.4.0 

76 Added support for optional positional arguments. 

77 """ 

78 return log 

79 

80 

81_FAIL_TYPES = (BaseException, Failure) 

82 

83 

84def _extractStuffAndWhy(eventDict: EventDict) -> tuple[Any, Any, EventDict]: 

85 """ 

86 Removes all possible *_why*s and *_stuff*s, analyzes exc_info and returns 

87 a tuple of ``(_stuff, _why, eventDict)``. 

88 

89 **Modifies** *eventDict*! 

90 """ 

91 _stuff = eventDict.pop("_stuff", None) 

92 _why = eventDict.pop("_why", None) 

93 event = eventDict.pop("event", None) 

94 

95 if isinstance(_stuff, _FAIL_TYPES) and isinstance(event, _FAIL_TYPES): 

96 raise ValueError("Both _stuff and event contain an Exception/Failure.") 

97 

98 # `log.err('event', _why='alsoEvent')` is ambiguous. 

99 if _why and isinstance(event, str): 

100 raise ValueError("Both `_why` and `event` supplied.") 

101 

102 # Two failures are ambiguous too. 

103 if not isinstance(_stuff, _FAIL_TYPES) and isinstance(event, _FAIL_TYPES): 

104 _why = _why or "error" 

105 _stuff = event 

106 

107 if isinstance(event, str): 

108 _why = event 

109 

110 if not _stuff and sys.exc_info() != (None, None, None): 

111 _stuff = Failure() # type: ignore[no-untyped-call] 

112 

113 # Either we used the error ourselves or the user supplied one for 

114 # formatting. Avoid log.err() to dump another traceback into the log. 

115 if isinstance(_stuff, BaseException) and not isinstance(_stuff, Failure): 

116 _stuff = Failure(_stuff) # type: ignore[no-untyped-call] 

117 

118 return _stuff, _why, eventDict 

119 

120 

121class ReprWrapper: 

122 """ 

123 Wrap a string and return it as the ``__repr__``. 

124 

125 This is needed for ``twisted.python.log.err`` that calls `repr` on 

126 ``_stuff``: 

127 

128 >>> repr("foo") 

129 "'foo'" 

130 >>> repr(ReprWrapper("foo")) 

131 'foo' 

132 

133 Note the extra quotes in the unwrapped example. 

134 """ 

135 

136 def __init__(self, string: str) -> None: 

137 self.string = string 

138 

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

140 """ 

141 Check for equality, just for tests. 

142 """ 

143 return ( 

144 isinstance(other, self.__class__) and self.string == other.string 

145 ) 

146 

147 def __repr__(self) -> str: 

148 return self.string 

149 

150 

151class JSONRenderer(GenericJSONRenderer): 

152 """ 

153 Behaves like `structlog.processors.JSONRenderer` except that it formats 

154 tracebacks and failures itself if called with ``err()``. 

155 

156 .. note:: 

157 

158 This ultimately means that the messages get logged out using ``msg()``, 

159 and *not* ``err()`` which renders failures in separate lines. 

160 

161 Therefore it will break your tests that contain assertions using 

162 `flushLoggedErrors 

163 <https://docs.twisted.org/en/stable/api/ 

164 twisted.trial.unittest.SynchronousTestCase.html#flushLoggedErrors>`_. 

165 

166 *Not* an adapter like `EventAdapter` but a real formatter. Also does *not* 

167 require to be adapted using it. 

168 

169 Use together with a `JSONLogObserverWrapper`-wrapped Twisted logger like 

170 `plainJSONStdOutLogger` for pure-JSON logs. 

171 """ 

172 

173 def __call__( # type: ignore[override] 

174 self, 

175 logger: WrappedLogger, 

176 name: str, 

177 eventDict: EventDict, 

178 ) -> tuple[Sequence[Any], dict[str, Any]]: 

179 _stuff, _why, eventDict = _extractStuffAndWhy(eventDict) 

180 if name == "err": 

181 eventDict["event"] = _why 

182 if isinstance(_stuff, Failure): 

183 eventDict["exception"] = _stuff.getTraceback(detail="verbose") 

184 _stuff.cleanFailure() # type: ignore[no-untyped-call] 

185 else: 

186 eventDict["event"] = _why 

187 return ( 

188 ( 

189 ReprWrapper( 

190 GenericJSONRenderer.__call__( # type: ignore[arg-type] 

191 self, logger, name, eventDict 

192 ) 

193 ), 

194 ), 

195 {"_structlog": True}, 

196 ) 

197 

198 

199@implementer(ILogObserver) 

200class PlainFileLogObserver: 

201 """ 

202 Write only the plain message without timestamps or anything else. 

203 

204 Great to just print JSON to stdout where you catch it with something like 

205 runit. 

206 

207 Args: 

208 file: File to print to. 

209 

210 .. versionadded:: 0.2.0 

211 """ 

212 

213 def __init__(self, file: TextIO) -> None: 

214 self._write = file.write 

215 self._flush = file.flush 

216 

217 def __call__(self, eventDict: EventDict) -> None: 

218 self._write( 

219 textFromEventDict(eventDict) # type: ignore[arg-type, operator] 

220 + "\n", 

221 ) 

222 self._flush() 

223 

224 

225@implementer(ILogObserver) 

226class JSONLogObserverWrapper: 

227 """ 

228 Wrap a log *observer* and render non-`JSONRenderer` entries to JSON. 

229 

230 Args: 

231 observer (ILogObserver): 

232 Twisted log observer to wrap. For example 

233 :class:`PlainFileObserver` or Twisted's stock `FileLogObserver 

234 <https://docs.twisted.org/en/stable/api/ 

235 twisted.python.log.FileLogObserver.html>`_ 

236 

237 .. versionadded:: 0.2.0 

238 """ 

239 

240 def __init__(self, observer: Any) -> None: 

241 self._observer = observer 

242 

243 def __call__(self, eventDict: EventDict) -> str: 

244 if "_structlog" not in eventDict: 

245 eventDict["message"] = ( 

246 json.dumps( 

247 { 

248 "event": textFromEventDict( 

249 eventDict # type: ignore[arg-type] 

250 ), 

251 "system": eventDict.get("system"), 

252 } 

253 ), 

254 ) 

255 eventDict["_structlog"] = True 

256 

257 return self._observer(eventDict) 

258 

259 

260def plainJSONStdOutLogger() -> JSONLogObserverWrapper: 

261 """ 

262 Return a logger that writes only the message to stdout. 

263 

264 Transforms non-`JSONRenderer` messages to JSON. 

265 

266 Ideal for JSONifying log entries from Twisted plugins and libraries that 

267 are outside of your control:: 

268 

269 $ twistd -n --logger structlog.twisted.plainJSONStdOutLogger web 

270 {"event": "Log opened.", "system": "-"} 

271 {"event": "twistd 13.1.0 (python 2.7.3) starting up.", "system": "-"} 

272 {"event": "reactor class: twisted...EPollReactor.", "system": "-"} 

273 {"event": "Site starting on 8080", "system": "-"} 

274 {"event": "Starting factory <twisted.web.server.Site ...>", ...} 

275 ... 

276 

277 Composes `PlainFileLogObserver` and `JSONLogObserverWrapper` to a usable 

278 logger. 

279 

280 .. versionadded:: 0.2.0 

281 """ 

282 return JSONLogObserverWrapper(PlainFileLogObserver(sys.stdout)) 

283 

284 

285class EventAdapter: 

286 """ 

287 Adapt an ``event_dict`` to Twisted logging system. 

288 

289 Particularly, make a wrapped `twisted.python.log.err 

290 <https://docs.twisted.org/en/stable/api/twisted.python.log.html#err>`_ 

291 behave as expected. 

292 

293 Args: 

294 dictRenderer: 

295 Renderer that is used for the actual log message. Please note that 

296 structlog comes with a dedicated `JSONRenderer`. 

297 

298 **Must** be the last processor in the chain and requires a *dictRenderer* 

299 for the actual formatting as an constructor argument in order to be able to 

300 fully support the original behaviors of ``log.msg()`` and ``log.err()``. 

301 """ 

302 

303 def __init__( 

304 self, 

305 dictRenderer: ( 

306 Callable[[WrappedLogger, str, EventDict], str] | None 

307 ) = None, 

308 ) -> None: 

309 self._dictRenderer = dictRenderer or _BUILTIN_DEFAULT_PROCESSORS[-1] 

310 

311 def __call__( 

312 self, logger: WrappedLogger, name: str, eventDict: EventDict 

313 ) -> Any: 

314 if name == "err": 

315 # This aspires to handle the following cases correctly: 

316 # 1. log.err(failure, _why='event', **kw) 

317 # 2. log.err('event', **kw) 

318 # 3. log.err(_stuff=failure, _why='event', **kw) 

319 _stuff, _why, eventDict = _extractStuffAndWhy(eventDict) 

320 eventDict["event"] = _why 

321 

322 return ( 

323 (), 

324 { 

325 "_stuff": _stuff, 

326 "_why": self._dictRenderer(logger, name, eventDict), 

327 }, 

328 ) 

329 

330 return self._dictRenderer(logger, name, eventDict)