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

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 classes responsible for output. 

8""" 

9 

10from __future__ import annotations 

11 

12import copy 

13import sys 

14import threading 

15import weakref 

16 

17from pickle import PicklingError 

18from sys import stderr, stdout 

19from typing import IO, Any, BinaryIO, TextIO 

20 

21 

22WRITE_LOCKS: weakref.WeakKeyDictionary[IO[Any], threading.Lock] = ( 

23 weakref.WeakKeyDictionary() 

24) 

25 

26 

27def _get_lock_for_file(file: IO[Any]) -> threading.Lock: 

28 lock = WRITE_LOCKS.get(file) 

29 if lock is None: 

30 lock = threading.Lock() 

31 WRITE_LOCKS[file] = lock 

32 

33 return lock 

34 

35 

36class PrintLogger: 

37 """ 

38 Print events into a file. 

39 

40 Args: 

41 file: File to print to. (default: `sys.stdout`) 

42 

43 >>> from structlog import PrintLogger 

44 >>> PrintLogger().info("hello") 

45 hello 

46 

47 Useful if you follow `current logging best practices 

48 <logging-best-practices>`. 

49 

50 Also very useful for testing and examples since `logging` is finicky in 

51 doctests. 

52 

53 .. versionchanged:: 22.1.0 

54 The implementation has been switched to use `print` for better 

55 monkeypatchability. 

56 """ 

57 

58 def __init__(self, file: TextIO | None = None): 

59 self._file = file or stdout 

60 

61 self._lock = _get_lock_for_file(self._file) 

62 

63 def __getstate__(self) -> str: 

64 """ 

65 Our __getattr__ magic makes this necessary. 

66 """ 

67 if self._file is stdout: 

68 return "stdout" 

69 

70 if self._file is stderr: 

71 return "stderr" 

72 

73 raise PicklingError( 

74 "Only PrintLoggers to sys.stdout and sys.stderr can be pickled." 

75 ) 

76 

77 def __setstate__(self, state: Any) -> None: 

78 """ 

79 Our __getattr__ magic makes this necessary. 

80 """ 

81 if state == "stdout": 

82 self._file = stdout 

83 else: 

84 self._file = stderr 

85 

86 self._lock = _get_lock_for_file(self._file) 

87 

88 def __deepcopy__(self, memodict: dict[str, object]) -> PrintLogger: 

89 """ 

90 Create a new PrintLogger with the same attributes. Similar to pickling. 

91 """ 

92 if self._file not in (stdout, stderr): 

93 raise copy.error( 

94 "Only PrintLoggers to sys.stdout and sys.stderr " 

95 "can be deepcopied." 

96 ) 

97 

98 newself = self.__class__(self._file) 

99 

100 newself._lock = _get_lock_for_file(newself._file) 

101 

102 return newself 

103 

104 def __repr__(self) -> str: 

105 return f"<PrintLogger(file={self._file!r})>" 

106 

107 def msg(self, message: str) -> None: 

108 """ 

109 Print *message*. 

110 """ 

111 f = self._file if self._file is not stdout else None 

112 with self._lock: 

113 print(message, file=f, flush=True) 

114 

115 log = debug = info = warn = warning = msg 

116 fatal = failure = err = error = critical = exception = msg 

117 

118 

119class PrintLoggerFactory: 

120 r""" 

121 Produce `PrintLogger`\ s. 

122 

123 To be used with `structlog.configure`\ 's ``logger_factory``. 

124 

125 Args: 

126 file: File to print to. (default: `sys.stdout`) 

127 

128 Positional arguments are silently ignored. 

129 

130 .. versionadded:: 0.4.0 

131 """ 

132 

133 def __init__(self, file: TextIO | None = None): 

134 self._file = file 

135 

136 def __call__(self, *args: Any) -> PrintLogger: 

137 return PrintLogger(self._file) 

138 

139 

140class WriteLogger: 

141 """ 

142 Write events into a file. 

143 

144 Args: 

145 file: File to print to. (default: `sys.stdout`) 

146 

147 >>> from structlog import WriteLogger 

148 >>> WriteLogger().info("hello") 

149 hello 

150 

151 Useful if you follow 

152 `current logging best practices <logging-best-practices>`. 

153 

154 Also very useful for testing and examples since `logging` is finicky in 

155 doctests. 

156 

157 A little faster and a little less versatile than `structlog.PrintLogger`. 

158 

159 .. versionadded:: 22.1.0 

160 """ 

161 

162 def __init__(self, file: TextIO | None = None): 

163 self._file = file or sys.stdout 

164 self._write = self._file.write 

165 self._flush = self._file.flush 

166 

167 self._lock = _get_lock_for_file(self._file) 

168 

169 def __getstate__(self) -> str: 

170 """ 

171 Our __getattr__ magic makes this necessary. 

172 """ 

173 if self._file is stdout: 

174 return "stdout" 

175 

176 if self._file is stderr: 

177 return "stderr" 

178 

179 raise PicklingError( 

180 "Only WriteLoggers to sys.stdout and sys.stderr can be pickled." 

181 ) 

182 

183 def __setstate__(self, state: Any) -> None: 

184 """ 

185 Our __getattr__ magic makes this necessary. 

186 """ 

187 if state == "stdout": 

188 self._file = stdout 

189 else: 

190 self._file = stderr 

191 

192 self._write = self._file.write 

193 self._flush = self._file.flush 

194 self._lock = _get_lock_for_file(self._file) 

195 

196 def __deepcopy__(self, memodict: dict[str, object]) -> WriteLogger: 

197 """ 

198 Create a new WriteLogger with the same attributes. Similar to pickling. 

199 """ 

200 if self._file not in (sys.stdout, sys.stderr): 

201 raise copy.error( 

202 "Only WriteLoggers to sys.stdout and sys.stderr " 

203 "can be deepcopied." 

204 ) 

205 

206 newself = self.__class__(self._file) 

207 

208 newself._write = newself._file.write 

209 newself._flush = newself._file.flush 

210 newself._lock = _get_lock_for_file(newself._file) 

211 

212 return newself 

213 

214 def __repr__(self) -> str: 

215 return f"<WriteLogger(file={self._file!r})>" 

216 

217 def msg(self, message: str) -> None: 

218 """ 

219 Write and flush *message*. 

220 """ 

221 with self._lock: 

222 self._write(message + "\n") 

223 self._flush() 

224 

225 log = debug = info = warn = warning = msg 

226 fatal = failure = err = error = critical = exception = msg 

227 

228 

229class WriteLoggerFactory: 

230 r""" 

231 Produce `WriteLogger`\ s. 

232 

233 To be used with `structlog.configure`\ 's ``logger_factory``. 

234 

235 Args: 

236 file: File to print to. (default: `sys.stdout`) 

237 

238 Positional arguments are silently ignored. 

239 

240 .. versionadded:: 22.1.0 

241 """ 

242 

243 def __init__(self, file: TextIO | None = None): 

244 self._file = file 

245 

246 def __call__(self, *args: Any) -> WriteLogger: 

247 return WriteLogger(self._file) 

248 

249 

250class BytesLogger: 

251 r""" 

252 Writes bytes into a file. 

253 

254 Useful if you follow `current logging best practices 

255 <logging-best-practices>` together with a formatter that returns bytes 

256 (e.g. `orjson <https://github.com/ijl/orjson>`_). 

257 

258 Args: 

259 file: File to print to. (default: `sys.stdout`\ ``.buffer``) 

260 

261 name: 

262 Optional name for the logger. If provided, it will be picked up 

263 as the logger's name when used with 

264 `structlog.stdlib.add_logger_name()` without using standard 

265 library integration. ``BytesLogger`` itself does nothing with it. 

266 

267 .. versionadded:: 20.2.0 

268 

269 .. versionadded:: 26.1.0 The ``name`` attribute. 

270 """ 

271 

272 __slots__ = ("_file", "_flush", "_lock", "_write", "name") 

273 

274 def __init__( 

275 self, file: BinaryIO | None = None, *, name: str | None = None 

276 ): 

277 self._file = file or sys.stdout.buffer 

278 self._write = self._file.write 

279 self._flush = self._file.flush 

280 

281 self.name = name 

282 

283 self._lock = _get_lock_for_file(self._file) 

284 

285 def __getstate__(self) -> tuple[str, str | None]: 

286 """ 

287 Our __getattr__ magic makes this necessary. 

288 """ 

289 if self._file is sys.stdout.buffer: 

290 return "stdout", self.name 

291 

292 if self._file is sys.stderr.buffer: 

293 return "stderr", self.name 

294 

295 raise PicklingError( 

296 "Only BytesLoggers to sys.stdout and sys.stderr can be pickled." 

297 ) 

298 

299 def __setstate__(self, state: Any) -> None: 

300 """ 

301 Our __getattr__ magic makes this necessary. 

302 """ 

303 if isinstance(state, str): 

304 name = None 

305 else: 

306 state, name = state 

307 

308 if state == "stdout": 

309 self._file = sys.stdout.buffer 

310 else: 

311 self._file = sys.stderr.buffer 

312 

313 self._write = self._file.write 

314 self._flush = self._file.flush 

315 self.name = name 

316 self._lock = _get_lock_for_file(self._file) 

317 

318 def __deepcopy__(self, memodict: dict[str, object]) -> BytesLogger: 

319 """ 

320 Create a new BytesLogger with the same attributes. Similar to pickling. 

321 """ 

322 if self._file not in (sys.stdout.buffer, sys.stderr.buffer): 

323 raise copy.error( 

324 "Only BytesLoggers to sys.stdout and sys.stderr " 

325 "can be deepcopied." 

326 ) 

327 

328 newself = self.__class__(self._file, name=self.name) 

329 

330 newself._write = newself._file.write 

331 newself._flush = newself._file.flush 

332 newself._lock = _get_lock_for_file(newself._file) 

333 

334 return newself 

335 

336 def __repr__(self) -> str: 

337 if self.name is None: 

338 return f"<BytesLogger(file={self._file!r})>" 

339 

340 return f"<BytesLogger(name={self.name!r}, file={self._file!r})>" 

341 

342 def msg(self, message: bytes) -> None: 

343 """ 

344 Write *message*. 

345 """ 

346 with self._lock: 

347 self._write(message + b"\n") 

348 self._flush() 

349 

350 log = debug = info = warn = warning = msg 

351 fatal = failure = err = error = critical = exception = msg 

352 

353 

354class BytesLoggerFactory: 

355 r""" 

356 Produce `BytesLogger`\ s. 

357 

358 To be used with `structlog.configure`\ 's ``logger_factory``. 

359 

360 Args: 

361 file: File to print to. (default: `sys.stdout`\ ``.buffer``) 

362 

363 Positional arguments are silently ignored. 

364 

365 .. versionadded:: 20.2.0 

366 """ 

367 

368 __slots__ = ("_file",) 

369 

370 def __init__(self, file: BinaryIO | None = None): 

371 self._file = file 

372 

373 def __call__(self, *args: Any) -> BytesLogger: 

374 return BytesLogger(self._file, name=args[0] if args else None)