Coverage for stlog/setup.py: 75%

91 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2025-06-26 07:52 +0000

1from __future__ import annotations 

2 

3import logging 

4import os 

5import sys 

6import traceback 

7import types 

8import typing 

9import warnings 

10 

11from stlog.adapter import getLogger 

12from stlog.base import GLOBAL_LOGGING_CONFIG, check_env_false 

13from stlog.formatter import ( 

14 DEFAULT_STLOG_GCP_JSON_FORMAT, 

15 JsonFormatter, 

16) 

17from stlog.output import Output, StreamOutput, make_stream_or_rich_stream_output 

18 

19DEFAULT_LEVEL: str = os.environ.get("STLOG_LEVEL", "INFO") 

20DEFAULT_CAPTURE_WARNINGS: bool = check_env_false("STLOG_CAPTURE_WARNINGS", True) 

21DEFAULT_PROGRAM_NAME: str | None = os.environ.get("STLOG_PROGRAM_NAME", None) 

22DEFAULT_DESTINATION: str = os.environ.get("STLOG_DESTINATION", "stderr").lower() 

23DEFAULT_OUTPUT: str = os.environ.get("STLOG_OUTPUT", "console").lower() 

24 

25 

26def _make_default_stream() -> typing.TextIO: 

27 if DEFAULT_DESTINATION == "stderr": 

28 return sys.stderr 

29 elif DEFAULT_DESTINATION == "stdout": 

30 return sys.stdout 

31 raise Exception( 

32 f"bad value:{DEFAULT_DESTINATION} for STLOG_DESTINATION env var => must be 'stderr' or 'stdout'" 

33 ) 

34 

35 

36def _make_default_outputs() -> list[Output]: 

37 if DEFAULT_OUTPUT == "console": 

38 return [make_stream_or_rich_stream_output(stream=_make_default_stream())] 

39 elif DEFAULT_OUTPUT == "json": 

40 return [ 

41 StreamOutput( 

42 stream=_make_default_stream(), 

43 formatter=JsonFormatter(), 

44 ) 

45 ] 

46 elif DEFAULT_OUTPUT == "json-human": 

47 return [ 

48 StreamOutput( 

49 stream=_make_default_stream(), 

50 formatter=JsonFormatter(indent=4), 

51 ) 

52 ] 

53 elif DEFAULT_OUTPUT == "json-gcp": 

54 return [ 

55 StreamOutput( 

56 stream=_make_default_stream(), 

57 formatter=JsonFormatter( 

58 fmt=DEFAULT_STLOG_GCP_JSON_FORMAT, 

59 ), 

60 ) 

61 ] 

62 else: 

63 raise Exception( 

64 f"bad value:{DEFAULT_OUTPUT} for STLOG_OUTPUT env var => must be 'console', 'json', 'json-human' or 'json-gcp'" 

65 ) 

66 

67 

68def _logging_excepthook( 

69 exc_type: type[BaseException], 

70 value: BaseException, 

71 tb: types.TracebackType | None = None, 

72) -> None: 

73 if issubclass(exc_type, KeyboardInterrupt): 

74 sys.__excepthook__(exc_type, value, tb) 

75 return 

76 try: 

77 program_logger = getLogger(GLOBAL_LOGGING_CONFIG.program_name) 

78 program_logger.error( 

79 "Exception catched in excepthook", exc_info=(exc_type, value, tb) 

80 ) 

81 except Exception: 

82 print( 

83 "ERROR: Exception during exception handling => let's dump this on standard output" 

84 ) 

85 print(traceback.format_exc(), file=sys.stderr) 

86 

87 

88def setup( # noqa: PLR0913 

89 *, 

90 level: str | int = DEFAULT_LEVEL, 

91 outputs: typing.Iterable[Output] | None = None, 

92 program_name: str | None = DEFAULT_PROGRAM_NAME, 

93 capture_warnings: bool = DEFAULT_CAPTURE_WARNINGS, 

94 logging_excepthook: typing.Callable[ 

95 [type[BaseException], BaseException, types.TracebackType | None], 

96 typing.Any, 

97 ] 

98 | None = _logging_excepthook, 

99 extra_levels: typing.Mapping[str, str | int] = {}, 

100 reinject_context_in_standard_logging: bool | None = None, 

101 read_extra_kwargs_from_standard_logging: bool | None = None, 

102) -> None: 

103 """Set up the Python logging with stlog (globally). 

104 

105 This removes all existing handlers and 

106 sets up handlers/formatters/... for Python logging. 

107 

108 Args: 

109 level: the root log level as int or as a string representation (in uppercase) 

110 (see https://docs.python.org/3/library/logging.html#levels), the default 

111 value is read in STLOG_LEVEL env var or set to INFO (if not set). 

112 outputs: iterable of Output to log to. 

113 program_name: the name of the program, the default value is read in STLOG_PROGRAM_NAME 

114 env var or auto-detected if not set. 

115 capture_warnings: capture warnings from the `warnings` module. 

116 logging_excepthook: if not None, override sys.excepthook with the given callable 

117 See https://docs.python.org/3/library/sys.html#sys.excepthook for details. 

118 extra_levels: dict "logger name => log level" for quick override of 

119 the root log level (for some loggers). 

120 reinject_context_in_standard_logging: if True, reinject the LogContext 

121 in log record emitted with python standard loggers (note: can be overriden per `stlog.output.Output`, 

122 default to `STLOG_REINJECT_CONTEXT_IN_STANDARD_LOGGING` env var or True if not set). 

123 read_extra_kwargs_from_standard_logging: if try to reinject the extra kwargs from standard logging 

124 (note: can be overriden per `stlog.output.Output`, default to `STLOG_READ_EXTRA_KWARGS_FROM_STANDARD_LOGGING` env var 

125 or False if not set). 

126 

127 """ 

128 GLOBAL_LOGGING_CONFIG.reinject_context_in_standard_logging = ( 

129 reinject_context_in_standard_logging 

130 ) 

131 GLOBAL_LOGGING_CONFIG.read_extra_kwargs_from_standard_logging = ( 

132 read_extra_kwargs_from_standard_logging 

133 ) 

134 if program_name is not None: 

135 GLOBAL_LOGGING_CONFIG.program_name = program_name 

136 

137 if GLOBAL_LOGGING_CONFIG._unit_tests_mode: 

138 # remove all configured loggers 

139 for key in list(logging.Logger.manager.loggerDict.keys()): 

140 logging.Logger.manager.loggerDict.pop(key) 

141 

142 root_logger = logging.getLogger(None) 

143 # Remove all handlers 

144 for handler in list(root_logger.handlers): 

145 root_logger.removeHandler(handler) 

146 

147 # Add configured handlers 

148 if outputs is None: 

149 outputs = _make_default_outputs() 

150 for out in outputs: 

151 root_logger.addHandler(out.get_handler()) 

152 

153 root_logger.setLevel(level) 

154 

155 if logging_excepthook: 

156 sys.excepthook = logging_excepthook 

157 

158 if capture_warnings: 

159 if GLOBAL_LOGGING_CONFIG._unit_tests_mode: 

160 # to avoid the capture by pytest 

161 logging._warnings_showwarning = None # type: ignore 

162 logging.captureWarnings(True) 

163 

164 for lgger, lvel in extra_levels.items(): 

165 logging.getLogger(lgger).setLevel(lvel) 

166 

167 GLOBAL_LOGGING_CONFIG.setup = True 

168 

169 

170ROOT_LOGGER = getLogger("root") 

171 

172 

173def log(level: int, msg, *args, **kwargs): 

174 """Log a message with the given integer severity level' on the root logger. 

175 

176 `setup()` is automatically called with default arguments if not already done before. 

177 """ 

178 if not GLOBAL_LOGGING_CONFIG.setup: 

179 setup() 

180 ROOT_LOGGER.log(level, msg, *args, **kwargs) 

181 

182 

183def debug(msg, *args, **kwargs): 

184 """Log a message with severity 'DEBUG' on the root logger. 

185 

186 `setup()` is automatically called with default arguments if not already done before. 

187 """ 

188 log(logging.DEBUG, msg, *args, **kwargs) 

189 

190 

191def info(msg, *args, **kwargs): 

192 """Log a message with severity 'INFO' on the root logger. 

193 

194 `setup()` is automatically called with default arguments if not already done before. 

195 """ 

196 log(logging.INFO, msg, *args, **kwargs) 

197 

198 

199def warning(msg, *args, **kwargs): 

200 """Log a message with severity 'WARNING' on the root logger. 

201 

202 `setup()` is automatically called with default arguments if not already done before. 

203 """ 

204 log(logging.WARNING, msg, *args, **kwargs) 

205 

206 

207def warn(msg, *args, **kwargs): 

208 if ( 

209 not GLOBAL_LOGGING_CONFIG.setup 

210 ): # we do this here to be able to capture the next warning 

211 setup() 

212 warnings.warn( 

213 "The 'warn' function is deprecated, use 'warning' instead", 

214 DeprecationWarning, 

215 2, 

216 ) 

217 warning(msg, *args, **kwargs) 

218 

219 

220def error(msg, *args, **kwargs): 

221 """Log a message with severity 'ERROR' on the root logger. 

222 

223 `setup()` is automatically called with default arguments if not already done before. 

224 """ 

225 log(logging.ERROR, msg, *args, **kwargs) 

226 

227 

228def critical(msg, *args, **kwargs): 

229 """Log a message with severity 'CRITICAL' on the root logger. 

230 

231 `setup()` is automatically called with default arguments if not already done before. 

232 """ 

233 log(logging.CRITICAL, msg, *args, **kwargs) 

234 

235 

236fatal = critical 

237 

238 

239def exception(msg, *args, exc_info=True, **kwargs): 

240 """Log a message with severity 'ERROR' on the root logger with exception information. 

241 

242 `setup()` is automatically called with default arguments if not already done before. 

243 """ 

244 error(msg, *args, exc_info=exc_info, **kwargs)