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

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

7Global state department. Don't reload this module or everything breaks. 

8""" 

9 

10from __future__ import annotations 

11 

12import os 

13import sys 

14import warnings 

15 

16from typing import Any, Callable, Iterable, Sequence, Type, cast 

17 

18from ._native import make_filtering_bound_logger 

19from ._output import PrintLoggerFactory 

20from .contextvars import merge_contextvars 

21from .dev import ConsoleRenderer, _has_colors, set_exc_info 

22from .processors import StackInfoRenderer, TimeStamper, add_log_level 

23from .typing import BindableLogger, Context, Processor, WrappedLogger 

24 

25 

26""" 

27Any changes to these defaults must be reflected in: 

28 

29- `getting-started`. 

30- structlog.stdlib.recreate_defaults()'s docstring. 

31""" 

32_BUILTIN_DEFAULT_PROCESSORS: Sequence[Processor] = [ 

33 merge_contextvars, 

34 add_log_level, 

35 StackInfoRenderer(), 

36 set_exc_info, 

37 TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False), 

38 ConsoleRenderer( 

39 colors=os.environ.get("NO_COLOR", "") == "" 

40 and ( 

41 os.environ.get("FORCE_COLOR", "") != "" 

42 or ( 

43 _has_colors 

44 and sys.stdout is not None 

45 and hasattr(sys.stdout, "isatty") 

46 and sys.stdout.isatty() 

47 ) 

48 ) 

49 ), 

50] 

51_BUILTIN_DEFAULT_CONTEXT_CLASS = cast(Type[Context], dict) 

52_BUILTIN_DEFAULT_WRAPPER_CLASS = make_filtering_bound_logger(0) 

53_BUILTIN_DEFAULT_LOGGER_FACTORY = PrintLoggerFactory() 

54_BUILTIN_CACHE_LOGGER_ON_FIRST_USE = False 

55 

56 

57class _Configuration: 

58 """ 

59 Global defaults. 

60 """ 

61 

62 is_configured: bool = False 

63 default_processors: Iterable[Processor] = _BUILTIN_DEFAULT_PROCESSORS[:] 

64 default_context_class: type[Context] = _BUILTIN_DEFAULT_CONTEXT_CLASS 

65 default_wrapper_class: Any = _BUILTIN_DEFAULT_WRAPPER_CLASS 

66 logger_factory: Callable[..., WrappedLogger] = ( 

67 _BUILTIN_DEFAULT_LOGGER_FACTORY 

68 ) 

69 cache_logger_on_first_use: bool = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

70 

71 

72_CONFIG = _Configuration() 

73""" 

74Global defaults used when arguments to `wrap_logger` are omitted. 

75""" 

76 

77 

78def is_configured() -> bool: 

79 """ 

80 Return whether *structlog* has been configured. 

81 

82 If `False`, *structlog* is running with builtin defaults. 

83 

84 .. versionadded: 18.1.0 

85 """ 

86 return _CONFIG.is_configured 

87 

88 

89def get_config() -> dict[str, Any]: 

90 """ 

91 Get a dictionary with the current configuration. 

92 

93 .. note:: 

94 

95 Changes to the returned dictionary do *not* affect *structlog*. 

96 

97 .. versionadded: 18.1.0 

98 """ 

99 return { 

100 "processors": _CONFIG.default_processors, 

101 "context_class": _CONFIG.default_context_class, 

102 "wrapper_class": _CONFIG.default_wrapper_class, 

103 "logger_factory": _CONFIG.logger_factory, 

104 "cache_logger_on_first_use": _CONFIG.cache_logger_on_first_use, 

105 } 

106 

107 

108def get_logger(*args: Any, **initial_values: Any) -> Any: 

109 """ 

110 Convenience function that returns a logger according to configuration. 

111 

112 >>> from structlog import get_logger 

113 >>> log = get_logger(y=23) 

114 >>> log.info("hello", x=42) 

115 y=23 x=42 event='hello' 

116 

117 Args: 

118 args: 

119 *Optional* positional arguments that are passed unmodified to the 

120 logger factory. Therefore it depends on the factory what they 

121 mean. 

122 

123 initial_values: Values that are used to pre-populate your contexts. 

124 

125 Returns: 

126 A proxy that creates a correctly configured bound logger when 

127 necessary. The type of that bound logger depends on your configuration 

128 and is `structlog.BoundLogger` by default. 

129 

130 See `configuration` for details. 

131 

132 If you prefer CamelCase, there's an alias for your reading pleasure: 

133 `structlog.getLogger`. 

134 

135 .. versionadded:: 0.4.0 *args* 

136 """ 

137 return wrap_logger(None, logger_factory_args=args, **initial_values) 

138 

139 

140getLogger = get_logger # noqa: N816 

141""" 

142CamelCase alias for `structlog.get_logger`. 

143 

144This function is supposed to be in every source file -- we don't want it to 

145stick out like a sore thumb in frameworks like Twisted or Zope. 

146""" 

147 

148 

149def wrap_logger( 

150 logger: WrappedLogger | None, 

151 processors: Iterable[Processor] | None = None, 

152 wrapper_class: type[BindableLogger] | None = None, 

153 context_class: type[Context] | None = None, 

154 cache_logger_on_first_use: bool | None = None, 

155 logger_factory_args: Iterable[Any] | None = None, 

156 **initial_values: Any, 

157) -> Any: 

158 """ 

159 Create a new bound logger for an arbitrary *logger*. 

160 

161 Default values for *processors*, *wrapper_class*, and *context_class* can 

162 be set using `configure`. 

163 

164 If you set an attribute here, `configure` calls have *no* effect for the 

165 *respective* attribute. 

166 

167 In other words: selective overwriting of the defaults while keeping some 

168 *is* possible. 

169 

170 Args: 

171 initial_values: Values that are used to pre-populate your contexts. 

172 

173 logger_factory_args: 

174 Values that are passed unmodified as ``*logger_factory_args`` to 

175 the logger factory if not `None`. 

176 

177 Returns: 

178 A proxy that creates a correctly configured bound logger when 

179 necessary. 

180 

181 See `configure` for the meaning of the rest of the arguments. 

182 

183 .. versionadded:: 0.4.0 *logger_factory_args* 

184 """ 

185 return BoundLoggerLazyProxy( 

186 logger, 

187 wrapper_class=wrapper_class, 

188 processors=processors, 

189 context_class=context_class, 

190 cache_logger_on_first_use=cache_logger_on_first_use, 

191 initial_values=initial_values, 

192 logger_factory_args=logger_factory_args, 

193 ) 

194 

195 

196def configure( 

197 processors: Iterable[Processor] | None = None, 

198 wrapper_class: type[BindableLogger] | None = None, 

199 context_class: type[Context] | None = None, 

200 logger_factory: Callable[..., WrappedLogger] | None = None, 

201 cache_logger_on_first_use: bool | None = None, 

202) -> None: 

203 """ 

204 Configures the **global** defaults. 

205 

206 They are used if `wrap_logger` or `get_logger` are called without 

207 arguments. 

208 

209 Can be called several times, keeping an argument at `None` leaves it 

210 unchanged from the current setting. 

211 

212 After calling for the first time, `is_configured` starts returning `True`. 

213 

214 Use `reset_defaults` to undo your changes. 

215 

216 Args: 

217 processors: The processor chain. See :doc:`processors` for details. 

218 

219 wrapper_class: 

220 Class to use for wrapping loggers instead of 

221 `structlog.BoundLogger`. See `standard-library`, :doc:`twisted`, 

222 and `custom-wrappers`. 

223 

224 context_class: 

225 Class to be used for internal context keeping. The default is a 

226 `dict` and since dictionaries are ordered as of Python 3.6, there's 

227 few reasons to change this option. 

228 

229 logger_factory: 

230 Factory to be called to create a new logger that shall be wrapped. 

231 

232 cache_logger_on_first_use: 

233 `wrap_logger` doesn't return an actual wrapped logger but a proxy 

234 that assembles one when it's first used. If this option is set to 

235 `True`, this assembled logger is cached. See `performance`. 

236 

237 .. versionadded:: 0.3.0 *cache_logger_on_first_use* 

238 """ 

239 _CONFIG.is_configured = True 

240 

241 if processors is not None: 

242 _CONFIG.default_processors = processors 

243 if wrapper_class is not None: 

244 _CONFIG.default_wrapper_class = wrapper_class 

245 if context_class is not None: 

246 _CONFIG.default_context_class = context_class 

247 if logger_factory is not None: 

248 _CONFIG.logger_factory = logger_factory 

249 if cache_logger_on_first_use is not None: 

250 _CONFIG.cache_logger_on_first_use = cache_logger_on_first_use 

251 

252 

253def configure_once( 

254 processors: Iterable[Processor] | None = None, 

255 wrapper_class: type[BindableLogger] | None = None, 

256 context_class: type[Context] | None = None, 

257 logger_factory: Callable[..., WrappedLogger] | None = None, 

258 cache_logger_on_first_use: bool | None = None, 

259) -> None: 

260 """ 

261 Configures if structlog isn't configured yet. 

262 

263 It does *not* matter whether it was configured using `configure` or 

264 `configure_once` before. 

265 

266 Raises: 

267 RuntimeWarning: if repeated configuration is attempted. 

268 """ 

269 if not _CONFIG.is_configured: 

270 configure( 

271 processors=processors, 

272 wrapper_class=wrapper_class, 

273 context_class=context_class, 

274 logger_factory=logger_factory, 

275 cache_logger_on_first_use=cache_logger_on_first_use, 

276 ) 

277 else: 

278 warnings.warn( 

279 "Repeated configuration attempted.", RuntimeWarning, stacklevel=2 

280 ) 

281 

282 

283def reset_defaults() -> None: 

284 """ 

285 Resets global default values to builtin defaults. 

286 

287 `is_configured` starts returning `False` afterwards. 

288 """ 

289 _CONFIG.is_configured = False 

290 _CONFIG.default_processors = _BUILTIN_DEFAULT_PROCESSORS[:] 

291 _CONFIG.default_wrapper_class = _BUILTIN_DEFAULT_WRAPPER_CLASS 

292 _CONFIG.default_context_class = _BUILTIN_DEFAULT_CONTEXT_CLASS 

293 _CONFIG.logger_factory = _BUILTIN_DEFAULT_LOGGER_FACTORY 

294 _CONFIG.cache_logger_on_first_use = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

295 

296 

297class BoundLoggerLazyProxy: 

298 """ 

299 Instantiates a bound logger on first usage. 

300 

301 Takes both configuration and instantiation parameters into account. 

302 

303 The only points where a bound logger changes state are ``bind()``, 

304 ``unbind()``, and ``new()`` and that return the actual ``BoundLogger``. 

305 

306 If and only if configuration says so, that actual bound logger is cached on 

307 first usage. 

308 

309 .. versionchanged:: 0.4.0 Added support for *logger_factory_args*. 

310 """ 

311 

312 # fulfill BindableLogger protocol without carrying accidental state 

313 @property 

314 def _context(self) -> dict[str, str]: 

315 return self._initial_values 

316 

317 def __init__( 

318 self, 

319 logger: WrappedLogger | None, 

320 wrapper_class: type[BindableLogger] | None = None, 

321 processors: Iterable[Processor] | None = None, 

322 context_class: type[Context] | None = None, 

323 cache_logger_on_first_use: bool | None = None, 

324 initial_values: dict[str, Any] | None = None, 

325 logger_factory_args: Any = None, 

326 ) -> None: 

327 self._logger = logger 

328 self._wrapper_class = wrapper_class 

329 self._processors = processors 

330 self._context_class = context_class 

331 self._cache_logger_on_first_use = cache_logger_on_first_use 

332 self._initial_values = initial_values or {} 

333 self._logger_factory_args = logger_factory_args or () 

334 

335 def __repr__(self) -> str: 

336 return ( 

337 f"<BoundLoggerLazyProxy(logger={self._logger!r}, wrapper_class=" 

338 f"{self._wrapper_class!r}, processors={self._processors!r}, " 

339 f"context_class={self._context_class!r}, " 

340 f"initial_values={self._initial_values!r}, " 

341 f"logger_factory_args={self._logger_factory_args!r})>" 

342 ) 

343 

344 def bind(self, **new_values: Any) -> BindableLogger: 

345 """ 

346 Assemble a new BoundLogger from arguments and configuration. 

347 """ 

348 if self._context_class: 

349 ctx = self._context_class(self._initial_values) 

350 else: 

351 ctx = _CONFIG.default_context_class(self._initial_values) 

352 

353 _logger = self._logger 

354 if not _logger: 

355 _logger = _CONFIG.logger_factory(*self._logger_factory_args) 

356 

357 if self._processors is None: 

358 procs = _CONFIG.default_processors 

359 else: 

360 procs = self._processors 

361 

362 cls = self._wrapper_class or _CONFIG.default_wrapper_class 

363 # Looks like Protocols ignore definitions of __init__ so we have to 

364 # silence Mypy here. 

365 logger = cls( 

366 _logger, 

367 processors=procs, 

368 context=ctx, # type: ignore[call-arg] 

369 ) 

370 

371 def finalized_bind(**new_values: Any) -> BindableLogger: 

372 """ 

373 Use cached assembled logger to bind potentially new values. 

374 """ 

375 if new_values: 

376 return logger.bind(**new_values) 

377 

378 return logger 

379 

380 if self._cache_logger_on_first_use is True or ( 

381 self._cache_logger_on_first_use is None 

382 and _CONFIG.cache_logger_on_first_use is True 

383 ): 

384 self.bind = finalized_bind # type: ignore[method-assign] 

385 

386 return finalized_bind(**new_values) 

387 

388 def unbind(self, *keys: str) -> BindableLogger: 

389 """ 

390 Same as bind, except unbind *keys* first. 

391 

392 In our case that could be only initial values. 

393 """ 

394 return self.bind().unbind(*keys) 

395 

396 def try_unbind(self, *keys: str) -> BindableLogger: 

397 return self.bind().try_unbind(*keys) 

398 

399 def new(self, **new_values: Any) -> BindableLogger: 

400 """ 

401 Clear context, then bind. 

402 """ 

403 if self._context_class: 

404 self._context_class().clear() 

405 else: 

406 _CONFIG.default_context_class().clear() 

407 

408 return self.bind(**new_values) 

409 

410 def __getattr__(self, name: str) -> Any: 

411 """ 

412 If a logging method if called on a lazy proxy, we have to create an 

413 ephemeral BoundLogger first. 

414 """ 

415 if name == "__isabstractmethod__": 

416 raise AttributeError 

417 

418 bl = self.bind() 

419 

420 return getattr(bl, name) 

421 

422 def __getstate__(self) -> dict[str, Any]: 

423 """ 

424 Our __getattr__ magic makes this necessary. 

425 """ 

426 return self.__dict__ 

427 

428 def __setstate__(self, state: dict[str, Any]) -> None: 

429 """ 

430 Our __getattr__ magic makes this necessary. 

431 """ 

432 for k, v in state.items(): 

433 setattr(self, k, v)