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

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 

15 

16from pickle import PicklingError 

17from sys import stderr, stdout 

18from typing import IO, Any, BinaryIO, TextIO 

19 

20 

21WRITE_LOCKS: dict[IO[Any], threading.Lock] = {} 

22 

23 

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

25 lock = WRITE_LOCKS.get(file) 

26 if lock is None: 

27 lock = threading.Lock() 

28 WRITE_LOCKS[file] = lock 

29 

30 return lock 

31 

32 

33class PrintLogger: 

34 """ 

35 Print events into a file. 

36 

37 Args: 

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

39 

40 >>> from structlog import PrintLogger 

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

42 hello 

43 

44 Useful if you follow `current logging best practices 

45 <logging-best-practices>`. 

46 

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

48 doctests. 

49 

50 .. versionchanged:: 22.1.0 

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

52 monkeypatchability. 

53 """ 

54 

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

56 self._file = file or stdout 

57 

58 self._lock = _get_lock_for_file(self._file) 

59 

60 def __getstate__(self) -> str: 

61 """ 

62 Our __getattr__ magic makes this necessary. 

63 """ 

64 if self._file is stdout: 

65 return "stdout" 

66 

67 if self._file is stderr: 

68 return "stderr" 

69 

70 raise PicklingError( 

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

72 ) 

73 

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

75 """ 

76 Our __getattr__ magic makes this necessary. 

77 """ 

78 if state == "stdout": 

79 self._file = stdout 

80 else: 

81 self._file = stderr 

82 

83 self._lock = _get_lock_for_file(self._file) 

84 

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

86 """ 

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

88 """ 

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

90 raise copy.error( 

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

92 "can be deepcopied." 

93 ) 

94 

95 newself = self.__class__(self._file) 

96 

97 newself._lock = _get_lock_for_file(newself._file) 

98 

99 return newself 

100 

101 def __repr__(self) -> str: 

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

103 

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

105 """ 

106 Print *message*. 

107 """ 

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

109 with self._lock: 

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

111 

112 log = debug = info = warn = warning = msg 

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

114 

115 

116class PrintLoggerFactory: 

117 r""" 

118 Produce `PrintLogger`\ s. 

119 

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

121 

122 Args: 

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

124 

125 Positional arguments are silently ignored. 

126 

127 .. versionadded:: 0.4.0 

128 """ 

129 

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

131 self._file = file 

132 

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

134 return PrintLogger(self._file) 

135 

136 

137class WriteLogger: 

138 """ 

139 Write events into a file. 

140 

141 Args: 

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

143 

144 >>> from structlog import WriteLogger 

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

146 hello 

147 

148 Useful if you follow 

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

150 

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

152 doctests. 

153 

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

155 

156 .. versionadded:: 22.1.0 

157 """ 

158 

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

160 self._file = file or sys.stdout 

161 self._write = self._file.write 

162 self._flush = self._file.flush 

163 

164 self._lock = _get_lock_for_file(self._file) 

165 

166 def __getstate__(self) -> str: 

167 """ 

168 Our __getattr__ magic makes this necessary. 

169 """ 

170 if self._file is stdout: 

171 return "stdout" 

172 

173 if self._file is stderr: 

174 return "stderr" 

175 

176 raise PicklingError( 

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

178 ) 

179 

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

181 """ 

182 Our __getattr__ magic makes this necessary. 

183 """ 

184 if state == "stdout": 

185 self._file = stdout 

186 else: 

187 self._file = stderr 

188 

189 self._lock = _get_lock_for_file(self._file) 

190 

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

192 """ 

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

194 """ 

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

196 raise copy.error( 

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

198 "can be deepcopied." 

199 ) 

200 

201 newself = self.__class__(self._file) 

202 

203 newself._write = newself._file.write 

204 newself._flush = newself._file.flush 

205 newself._lock = _get_lock_for_file(newself._file) 

206 

207 return newself 

208 

209 def __repr__(self) -> str: 

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

211 

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

213 """ 

214 Write and flush *message*. 

215 """ 

216 with self._lock: 

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

218 self._flush() 

219 

220 log = debug = info = warn = warning = msg 

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

222 

223 

224class WriteLoggerFactory: 

225 r""" 

226 Produce `WriteLogger`\ s. 

227 

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

229 

230 Args: 

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

232 

233 Positional arguments are silently ignored. 

234 

235 .. versionadded:: 22.1.0 

236 """ 

237 

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

239 self._file = file 

240 

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

242 return WriteLogger(self._file) 

243 

244 

245class BytesLogger: 

246 r""" 

247 Writes bytes into a file. 

248 

249 Args: 

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

251 

252 Useful if you follow `current logging best practices 

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

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

255 

256 .. versionadded:: 20.2.0 

257 """ 

258 

259 __slots__ = ("_file", "_flush", "_lock", "_write") 

260 

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

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

263 self._write = self._file.write 

264 self._flush = self._file.flush 

265 

266 self._lock = _get_lock_for_file(self._file) 

267 

268 def __getstate__(self) -> str: 

269 """ 

270 Our __getattr__ magic makes this necessary. 

271 """ 

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

273 return "stdout" 

274 

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

276 return "stderr" 

277 

278 raise PicklingError( 

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

280 ) 

281 

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

283 """ 

284 Our __getattr__ magic makes this necessary. 

285 """ 

286 if state == "stdout": 

287 self._file = sys.stdout.buffer 

288 else: 

289 self._file = sys.stderr.buffer 

290 

291 self._write = self._file.write 

292 self._flush = self._file.flush 

293 self._lock = _get_lock_for_file(self._file) 

294 

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

296 """ 

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

298 """ 

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

300 raise copy.error( 

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

302 "can be deepcopied." 

303 ) 

304 

305 newself = self.__class__(self._file) 

306 

307 newself._write = newself._file.write 

308 newself._flush = newself._file.flush 

309 newself._lock = _get_lock_for_file(newself._file) 

310 

311 return newself 

312 

313 def __repr__(self) -> str: 

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

315 

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

317 """ 

318 Write *message*. 

319 """ 

320 with self._lock: 

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

322 self._flush() 

323 

324 log = debug = info = warn = warning = msg 

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

326 

327 

328class BytesLoggerFactory: 

329 r""" 

330 Produce `BytesLogger`\ s. 

331 

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

333 

334 Args: 

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

336 

337 Positional arguments are silently ignored. 

338 

339 .. versionadded:: 20.2.0 

340 """ 

341 

342 __slots__ = ("_file",) 

343 

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

345 self._file = file 

346 

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

348 return BytesLogger(self._file)