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 

33_no_colors = os.environ.get("NO_COLOR", "") != "" 

34_force_colors = os.environ.get("FORCE_COLOR", "") != "" 

35 

36_BUILTIN_DEFAULT_PROCESSORS: Sequence[Processor] = [ 

37 merge_contextvars, 

38 add_log_level, 

39 StackInfoRenderer(), 

40 set_exc_info, 

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

42 ConsoleRenderer( 

43 colors=not _no_colors 

44 and ( 

45 _force_colors 

46 or ( 

47 _has_colors 

48 and sys.stdout is not None 

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

50 and sys.stdout.isatty() 

51 ) 

52 ), 

53 force_colors=_force_colors, 

54 ), 

55] 

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

57_BUILTIN_DEFAULT_WRAPPER_CLASS = make_filtering_bound_logger(0) 

58_BUILTIN_DEFAULT_LOGGER_FACTORY = PrintLoggerFactory() 

59_BUILTIN_CACHE_LOGGER_ON_FIRST_USE = False 

60 

61 

62class _Configuration: 

63 """ 

64 Global defaults. 

65 """ 

66 

67 is_configured: bool = False 

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

69 default_context_class: type[Context] = _BUILTIN_DEFAULT_CONTEXT_CLASS 

70 default_wrapper_class: Any = _BUILTIN_DEFAULT_WRAPPER_CLASS 

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

72 _BUILTIN_DEFAULT_LOGGER_FACTORY 

73 ) 

74 cache_logger_on_first_use: bool = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

75 

76 

77_CONFIG = _Configuration() 

78""" 

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

80""" 

81 

82 

83def is_configured() -> bool: 

84 """ 

85 Return whether *structlog* has been configured. 

86 

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

88 

89 .. versionadded: 18.1.0 

90 """ 

91 return _CONFIG.is_configured 

92 

93 

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

95 """ 

96 Get a dictionary with the current configuration. 

97 

98 .. note:: 

99 

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

101 

102 .. versionadded: 18.1.0 

103 """ 

104 return { 

105 "processors": _CONFIG.default_processors, 

106 "context_class": _CONFIG.default_context_class, 

107 "wrapper_class": _CONFIG.default_wrapper_class, 

108 "logger_factory": _CONFIG.logger_factory, 

109 "cache_logger_on_first_use": _CONFIG.cache_logger_on_first_use, 

110 } 

111 

112 

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

114 """ 

115 Convenience function that returns a logger according to configuration. 

116 

117 >>> from structlog import get_logger 

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

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

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

121 

122 Args: 

123 args: 

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

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

126 mean. 

127 

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

129 

130 Returns: 

131 A proxy that creates a correctly configured bound logger when 

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

133 and is `structlog.BoundLogger` by default. 

134 

135 See `configuration` for details. 

136 

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

138 `structlog.getLogger`. 

139 

140 .. versionadded:: 0.4.0 *args* 

141 """ 

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

143 

144 

145getLogger = get_logger # noqa: N816 

146""" 

147CamelCase alias for `structlog.get_logger`. 

148 

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

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

151""" 

152 

153 

154def wrap_logger( 

155 logger: WrappedLogger | None, 

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

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

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

159 cache_logger_on_first_use: bool | None = None, 

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

161 **initial_values: Any, 

162) -> Any: 

163 """ 

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

165 

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

167 be set using `configure`. 

168 

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

170 *respective* attribute. 

171 

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

173 *is* possible. 

174 

175 Args: 

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

177 

178 logger_factory_args: 

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

180 the logger factory if not `None`. 

181 

182 Returns: 

183 A proxy that creates a correctly configured bound logger when 

184 necessary. 

185 

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

187 

188 .. versionadded:: 0.4.0 *logger_factory_args* 

189 """ 

190 return BoundLoggerLazyProxy( 

191 logger, 

192 wrapper_class=wrapper_class, 

193 processors=processors, 

194 context_class=context_class, 

195 cache_logger_on_first_use=cache_logger_on_first_use, 

196 initial_values=initial_values, 

197 logger_factory_args=logger_factory_args, 

198 ) 

199 

200 

201def configure( 

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

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

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

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

206 cache_logger_on_first_use: bool | None = None, 

207) -> None: 

208 """ 

209 Configures the **global** defaults. 

210 

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

212 arguments. 

213 

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

215 unchanged from the current setting. 

216 

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

218 

219 Use `reset_defaults` to undo your changes. 

220 

221 Args: 

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

223 

224 wrapper_class: 

225 Class to use for wrapping loggers instead of 

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

227 and `custom-wrappers`. 

228 

229 context_class: 

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

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

232 few reasons to change this option. 

233 

234 logger_factory: 

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

236 

237 cache_logger_on_first_use: 

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

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

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

241 

242 .. versionadded:: 0.3.0 *cache_logger_on_first_use* 

243 """ 

244 _CONFIG.is_configured = True 

245 

246 if processors is not None: 

247 _CONFIG.default_processors = processors 

248 if wrapper_class is not None: 

249 _CONFIG.default_wrapper_class = wrapper_class 

250 if context_class is not None: 

251 _CONFIG.default_context_class = context_class 

252 if logger_factory is not None: 

253 _CONFIG.logger_factory = logger_factory 

254 if cache_logger_on_first_use is not None: 

255 _CONFIG.cache_logger_on_first_use = cache_logger_on_first_use 

256 

257 

258def configure_once( 

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

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

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

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

263 cache_logger_on_first_use: bool | None = None, 

264) -> None: 

265 """ 

266 Configures if structlog isn't configured yet. 

267 

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

269 `configure_once` before. 

270 

271 Raises: 

272 RuntimeWarning: if repeated configuration is attempted. 

273 """ 

274 if not _CONFIG.is_configured: 

275 configure( 

276 processors=processors, 

277 wrapper_class=wrapper_class, 

278 context_class=context_class, 

279 logger_factory=logger_factory, 

280 cache_logger_on_first_use=cache_logger_on_first_use, 

281 ) 

282 else: 

283 warnings.warn( 

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

285 ) 

286 

287 

288def reset_defaults() -> None: 

289 """ 

290 Resets global default values to builtin defaults. 

291 

292 `is_configured` starts returning `False` afterwards. 

293 """ 

294 _CONFIG.is_configured = False 

295 _CONFIG.default_processors = _BUILTIN_DEFAULT_PROCESSORS[:] 

296 _CONFIG.default_wrapper_class = _BUILTIN_DEFAULT_WRAPPER_CLASS 

297 _CONFIG.default_context_class = _BUILTIN_DEFAULT_CONTEXT_CLASS 

298 _CONFIG.logger_factory = _BUILTIN_DEFAULT_LOGGER_FACTORY 

299 _CONFIG.cache_logger_on_first_use = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

300 

301 

302class BoundLoggerLazyProxy: 

303 """ 

304 Instantiates a bound logger on first usage. 

305 

306 Takes both configuration and instantiation parameters into account. 

307 

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

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

310 

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

312 first usage. 

313 

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

315 """ 

316 

317 # fulfill BindableLogger protocol without carrying accidental state 

318 @property 

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

320 return self._initial_values 

321 

322 def __init__( 

323 self, 

324 logger: WrappedLogger | None, 

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

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

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

328 cache_logger_on_first_use: bool | None = None, 

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

330 logger_factory_args: Any = None, 

331 ) -> None: 

332 self._logger = logger 

333 self._wrapper_class = wrapper_class 

334 self._processors = processors 

335 self._context_class = context_class 

336 self._cache_logger_on_first_use = cache_logger_on_first_use 

337 self._initial_values = initial_values or {} 

338 self._logger_factory_args = logger_factory_args or () 

339 

340 def __repr__(self) -> str: 

341 return ( 

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

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

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

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

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

347 ) 

348 

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

350 """ 

351 Assemble a new BoundLogger from arguments and configuration. 

352 """ 

353 if self._context_class: 

354 ctx = self._context_class(self._initial_values) 

355 else: 

356 ctx = _CONFIG.default_context_class(self._initial_values) 

357 

358 _logger = self._logger 

359 if not _logger: 

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

361 

362 if self._processors is None: 

363 procs = _CONFIG.default_processors 

364 else: 

365 procs = self._processors 

366 

367 cls = self._wrapper_class or _CONFIG.default_wrapper_class 

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

369 # silence Mypy here. 

370 logger = cls( 

371 _logger, 

372 processors=procs, 

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

374 ) 

375 

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

377 """ 

378 Use cached assembled logger to bind potentially new values. 

379 """ 

380 if new_values: 

381 return logger.bind(**new_values) 

382 

383 return logger 

384 

385 if self._cache_logger_on_first_use is True or ( 

386 self._cache_logger_on_first_use is None 

387 and _CONFIG.cache_logger_on_first_use is True 

388 ): 

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

390 

391 return finalized_bind(**new_values) 

392 

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

394 """ 

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

396 

397 In our case that could be only initial values. 

398 """ 

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

400 

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

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

403 

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

405 """ 

406 Clear context, then bind. 

407 """ 

408 if self._context_class: 

409 self._context_class().clear() 

410 else: 

411 _CONFIG.default_context_class().clear() 

412 

413 return self.bind(**new_values) 

414 

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

416 """ 

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

418 ephemeral BoundLogger first. 

419 """ 

420 if name == "__isabstractmethod__": 

421 raise AttributeError 

422 

423 bl = self.bind() 

424 

425 return getattr(bl, name) 

426 

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

428 """ 

429 Our __getattr__ magic makes this necessary. 

430 """ 

431 return self.__dict__ 

432 

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

434 """ 

435 Our __getattr__ magic makes this necessary. 

436 """ 

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

438 setattr(self, k, v)