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 collections.abc import Callable, Iterable, Sequence 

17from typing import Any, cast 

18 

19from ._native import make_filtering_bound_logger 

20from ._output import PrintLoggerFactory 

21from .contextvars import merge_contextvars 

22from .dev import ConsoleRenderer, _has_colors, set_exc_info 

23from .processors import StackInfoRenderer, TimeStamper, add_log_level 

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

25 

26 

27""" 

28Any changes to these defaults must be reflected in: 

29 

30- `getting-started`. 

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

32""" 

33 

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

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

36 

37_BUILTIN_DEFAULT_PROCESSORS: Sequence[Processor] = [ 

38 merge_contextvars, 

39 add_log_level, 

40 StackInfoRenderer(), 

41 set_exc_info, 

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

43 ConsoleRenderer( 

44 colors=not _no_colors 

45 and ( 

46 _force_colors 

47 or ( 

48 _has_colors 

49 and sys.stdout is not None 

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

51 and sys.stdout.isatty() 

52 ) 

53 ), 

54 force_colors=_force_colors, 

55 ), 

56] 

57_BUILTIN_DEFAULT_CONTEXT_CLASS = cast(type[Context], dict) 

58_BUILTIN_DEFAULT_WRAPPER_CLASS = make_filtering_bound_logger(0) 

59_BUILTIN_DEFAULT_LOGGER_FACTORY = PrintLoggerFactory() 

60_BUILTIN_CACHE_LOGGER_ON_FIRST_USE = False 

61 

62 

63class _Configuration: 

64 """ 

65 Global defaults. 

66 """ 

67 

68 is_configured: bool = False 

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

70 default_context_class: type[Context] = _BUILTIN_DEFAULT_CONTEXT_CLASS 

71 default_wrapper_class: Any = _BUILTIN_DEFAULT_WRAPPER_CLASS 

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

73 _BUILTIN_DEFAULT_LOGGER_FACTORY 

74 ) 

75 cache_logger_on_first_use: bool = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

76 

77 

78_CONFIG = _Configuration() 

79""" 

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

81""" 

82 

83 

84def is_configured() -> bool: 

85 """ 

86 Return whether *structlog* has been configured. 

87 

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

89 

90 .. versionadded: 18.1.0 

91 """ 

92 return _CONFIG.is_configured 

93 

94 

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

96 """ 

97 Get a dictionary with the current configuration. 

98 

99 .. note:: 

100 

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

102 

103 .. versionadded: 18.1.0 

104 """ 

105 return { 

106 "processors": _CONFIG.default_processors, 

107 "context_class": _CONFIG.default_context_class, 

108 "wrapper_class": _CONFIG.default_wrapper_class, 

109 "logger_factory": _CONFIG.logger_factory, 

110 "cache_logger_on_first_use": _CONFIG.cache_logger_on_first_use, 

111 } 

112 

113 

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

115 """ 

116 Convenience function that returns a logger according to configuration. 

117 

118 >>> from structlog import get_logger 

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

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

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

122 

123 Args: 

124 args: 

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

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

127 mean. 

128 

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

130 

131 Returns: 

132 A proxy that creates a correctly configured bound logger when 

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

134 and is `structlog.BoundLogger` by default. 

135 

136 See `configuration` for details. 

137 

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

139 `structlog.getLogger`. 

140 

141 .. versionadded:: 0.4.0 *args* 

142 """ 

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

144 

145 

146getLogger = get_logger # noqa: N816 

147""" 

148CamelCase alias for `structlog.get_logger`. 

149 

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

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

152""" 

153 

154 

155def wrap_logger( 

156 logger: WrappedLogger | None, 

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

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

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

160 cache_logger_on_first_use: bool | None = None, 

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

162 **initial_values: Any, 

163) -> Any: 

164 """ 

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

166 

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

168 be set using `configure`. 

169 

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

171 *respective* attribute. 

172 

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

174 *is* possible. 

175 

176 Args: 

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

178 

179 logger_factory_args: 

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

181 the logger factory if not `None`. 

182 

183 Returns: 

184 A proxy that creates a correctly configured bound logger when 

185 necessary. 

186 

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

188 

189 .. versionadded:: 0.4.0 *logger_factory_args* 

190 """ 

191 return BoundLoggerLazyProxy( 

192 logger, 

193 wrapper_class=wrapper_class, 

194 processors=processors, 

195 context_class=context_class, 

196 cache_logger_on_first_use=cache_logger_on_first_use, 

197 initial_values=initial_values, 

198 logger_factory_args=logger_factory_args, 

199 ) 

200 

201 

202def configure( 

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

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

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

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

207 cache_logger_on_first_use: bool | None = None, 

208) -> None: 

209 """ 

210 Configures the **global** defaults. 

211 

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

213 arguments. 

214 

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

216 unchanged from the current setting. 

217 

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

219 

220 Use `reset_defaults` to undo your changes. 

221 

222 Args: 

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

224 

225 wrapper_class: 

226 Class to use for wrapping loggers instead of 

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

228 and `custom-wrappers`. 

229 

230 context_class: 

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

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

233 few reasons to change this option. 

234 

235 logger_factory: 

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

237 

238 cache_logger_on_first_use: 

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

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

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

242 

243 .. versionadded:: 0.3.0 *cache_logger_on_first_use* 

244 """ 

245 _CONFIG.is_configured = True 

246 

247 if processors is not None: 

248 _CONFIG.default_processors = processors 

249 if wrapper_class is not None: 

250 _CONFIG.default_wrapper_class = wrapper_class 

251 if context_class is not None: 

252 _CONFIG.default_context_class = context_class 

253 if logger_factory is not None: 

254 _CONFIG.logger_factory = logger_factory 

255 if cache_logger_on_first_use is not None: 

256 _CONFIG.cache_logger_on_first_use = cache_logger_on_first_use 

257 

258 

259def configure_once( 

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

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

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

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

264 cache_logger_on_first_use: bool | None = None, 

265) -> None: 

266 """ 

267 Configures if structlog isn't configured yet. 

268 

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

270 `configure_once` before. 

271 

272 Raises: 

273 RuntimeWarning: if repeated configuration is attempted. 

274 """ 

275 if not _CONFIG.is_configured: 

276 configure( 

277 processors=processors, 

278 wrapper_class=wrapper_class, 

279 context_class=context_class, 

280 logger_factory=logger_factory, 

281 cache_logger_on_first_use=cache_logger_on_first_use, 

282 ) 

283 else: 

284 warnings.warn( 

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

286 ) 

287 

288 

289def reset_defaults() -> None: 

290 """ 

291 Resets global default values to builtin defaults. 

292 

293 `is_configured` starts returning `False` afterwards. 

294 """ 

295 _CONFIG.is_configured = False 

296 _CONFIG.default_processors = _BUILTIN_DEFAULT_PROCESSORS[:] 

297 _CONFIG.default_wrapper_class = _BUILTIN_DEFAULT_WRAPPER_CLASS 

298 _CONFIG.default_context_class = _BUILTIN_DEFAULT_CONTEXT_CLASS 

299 _CONFIG.logger_factory = _BUILTIN_DEFAULT_LOGGER_FACTORY 

300 _CONFIG.cache_logger_on_first_use = _BUILTIN_CACHE_LOGGER_ON_FIRST_USE 

301 

302 

303class BoundLoggerLazyProxy: 

304 """ 

305 Instantiates a bound logger on first usage. 

306 

307 Takes both configuration and instantiation parameters into account. 

308 

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

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

311 

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

313 first usage. 

314 

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

316 """ 

317 

318 # fulfill BindableLogger protocol without carrying accidental state 

319 @property 

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

321 return self._initial_values 

322 

323 def __init__( 

324 self, 

325 logger: WrappedLogger | None, 

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

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

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

329 cache_logger_on_first_use: bool | None = None, 

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

331 logger_factory_args: Any = None, 

332 ) -> None: 

333 self._logger = logger 

334 self._wrapper_class = wrapper_class 

335 self._processors = processors 

336 self._context_class = context_class 

337 self._cache_logger_on_first_use = cache_logger_on_first_use 

338 self._initial_values = initial_values or {} 

339 self._logger_factory_args = logger_factory_args or () 

340 

341 def __repr__(self) -> str: 

342 return ( 

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

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

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

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

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

348 ) 

349 

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

351 """ 

352 Assemble a new BoundLogger from arguments and configuration. 

353 """ 

354 if self._context_class: 

355 ctx = self._context_class(self._initial_values) 

356 else: 

357 ctx = _CONFIG.default_context_class(self._initial_values) 

358 

359 _logger = self._logger 

360 if not _logger: 

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

362 

363 if self._processors is None: 

364 procs = _CONFIG.default_processors 

365 else: 

366 procs = self._processors 

367 

368 cls = self._wrapper_class or _CONFIG.default_wrapper_class 

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

370 # silence Mypy here. 

371 logger = cls( 

372 _logger, 

373 processors=procs, 

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

375 ) 

376 

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

378 """ 

379 Use cached assembled logger to bind potentially new values. 

380 """ 

381 if new_values: 

382 return logger.bind(**new_values) 

383 

384 return logger 

385 

386 if self._cache_logger_on_first_use is True or ( 

387 self._cache_logger_on_first_use is None 

388 and _CONFIG.cache_logger_on_first_use is True 

389 ): 

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

391 

392 return finalized_bind(**new_values) 

393 

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

395 """ 

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

397 

398 In our case that could be only initial values. 

399 """ 

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

401 

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

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

404 

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

406 """ 

407 Clear context, then bind. 

408 """ 

409 if self._context_class: 

410 self._context_class().clear() 

411 else: 

412 _CONFIG.default_context_class().clear() 

413 

414 return self.bind(**new_values) 

415 

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

417 """ 

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

419 ephemeral BoundLogger first. 

420 """ 

421 if name == "__isabstractmethod__": 

422 raise AttributeError 

423 

424 bl = self.bind() 

425 

426 return getattr(bl, name) 

427 

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

429 """ 

430 Our __getattr__ magic makes this necessary. 

431 """ 

432 return self.__dict__ 

433 

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

435 """ 

436 Our __getattr__ magic makes this necessary. 

437 """ 

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

439 setattr(self, k, v)