Coverage for gws-app/gws/core/log.py: 73%

1"""Minimal logging facility that writes to stdout.""" 

2 

3import os 

4import sys 

5import traceback 

6 

7 

8class Level: 

9 """Log level constants, as in the standard ``logging`` module.""" 

10 

11 CRITICAL = 50 

12 ERROR = 40 

13 WARN = 30 

14 WARNING = 30 

15 INFO = 20 

16 DEBUG = 10 

17 NOTSET = 0 

18 ALL = 0 

19 

20 

21def set_level(level: int | str | None): 

22 """Set the current log level. 

23 

24 Args: 

25 level: A numeric level, a level name like ``DEBUG``, or ``None`` for ``INFO``. 

26 """ 

27 global _current_level 

28 if level is None: 

29 _current_level = Level.INFO 

30 elif isinstance(level, int) or level.isdigit(): 

31 _current_level = int(level) 

32 else: 

33 _current_level = getattr(Level, level.upper()) 

34 

35 

36def get_level() -> str: 

37 """Return the name of the current log level. 

38 

39 Returns: 

40 A level name like ``INFO``. 

41 """ 

42 global _current_level 

43 for k, n in vars(Level).items(): 

44 if n == _current_level: 

45 return k 

46 return 'ALL' 

47 

48 

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

50 """Log a message at the given level. 

51 

52 Args: 

53 level: Numeric log level. 

54 msg: Message, optionally with ``%`` placeholders. 

55 *args: Values for the placeholders. A single dict is used as a mapping. 

56 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

57 """ 

58 _raw(level, msg, args, kwargs) 

59 

60 

61def critical(msg: str, *args, **kwargs): 

62 """Log a message at the critical level. 

63 

64 Args: 

65 msg: Message, optionally with ``%`` placeholders. 

66 *args: Values for the placeholders. A single dict is used as a mapping. 

67 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

68 """ 

69 _raw(Level.CRITICAL, msg, args, kwargs) 

70 

71 

72def error(msg: str, *args, **kwargs): 

73 """Log a message at the error level. 

74 

75 Args: 

76 msg: Message, optionally with ``%`` placeholders. 

77 *args: Values for the placeholders. A single dict is used as a mapping. 

78 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

79 """ 

80 _raw(Level.ERROR, msg, args, kwargs) 

81 

82 

83def warning(msg: str, *args, **kwargs): 

84 """Log a message at the warning level. 

85 

86 Args: 

87 msg: Message, optionally with ``%`` placeholders. 

88 *args: Values for the placeholders. A single dict is used as a mapping. 

89 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

90 """ 

91 _raw(Level.WARNING, msg, args, kwargs) 

92 

93 

94def info(msg: str, *args, **kwargs): 

95 """Log a message at the info level. 

96 

97 Args: 

98 msg: Message, optionally with ``%`` placeholders. 

99 *args: Values for the placeholders. A single dict is used as a mapping. 

100 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

101 """ 

102 _raw(Level.INFO, msg, args, kwargs) 

103 

104 

105def debug(msg: str, *args, **kwargs): 

106 """Log a message at the debug level. 

107 

108 Args: 

109 msg: Message, optionally with ``%`` placeholders. 

110 *args: Values for the placeholders. A single dict is used as a mapping. 

111 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

112 """ 

113 _raw(Level.DEBUG, msg, args, kwargs) 

114 

115 

116def exception(msg: str = '', *args, **kwargs): 

117 """Log the exception being handled, with its backtrace, at the error level. 

118 

119 Args: 

120 msg: Message, optionally with ``%`` placeholders. Defaults to the first line of the backtrace. 

121 *args: Values for the placeholders. A single dict is used as a mapping. 

122 **kwargs: ``stacklevel`` to adjust the reported source location in debug mode. 

123 """ 

124 _, exc, _ = sys.exc_info() 

125 if exc is None: 

126 _raw(Level.ERROR, msg or 'exception() called without an active exception', args, kwargs) 

127 return 

128 ls = exception_backtrace(exc) 

129 _raw(Level.ERROR, msg or ls[0], args, kwargs) 

130 for s in ls[1:]: 

131 _raw(Level.ERROR, 'EXCEPTION :: ' + s) 

132 

133 

134def if_debug(fn, *args): 

135 """If the log level is debug, apply the function to args and log the result. 

136 

137 Exceptions raised by the function are logged instead of the result. 

138 

139 Args: 

140 fn: A function that returns a message. 

141 *args: Arguments for the function. 

142 """ 

143 

144 if Level.DEBUG < _current_level: 

145 return 

146 try: 

147 msg = fn(*args) 

148 except Exception as exc: 

149 msg = repr(exc) 

150 _raw(Level.DEBUG, msg) 

151 

152 

153def exception_backtrace(exc: BaseException | None) -> list: 

154 """Return the exception backtrace as a list of strings. 

155 

156 The backtrace includes the chain of causes and contexts. 

157 

158 Args: 

159 exc: An exception. 

160 

161 Returns: 

162 A list of lines. The first line is a summary with the exception name, the first message and the location. 

163 """ 

164 

165 head = _name(exc) 

166 messages = [] 

167 

168 lines = [] 

169 pfx = '' 

170 

171 while exc: 

172 subhead = _name(exc) 

173 msg = _message(exc) 

174 if msg: 

175 subhead += ': ' + msg 

176 messages.append(msg) 

177 if pfx: 

178 subhead = pfx + ' ' + subhead 

179 

180 lines.append(subhead) 

181 

182 for f in traceback.extract_tb(exc.__traceback__, limit=100): 

183 lines.append(f' in {f[2]} ({f[0]}:{f[1]})') 

184 

185 if exc.__cause__: 

186 exc = exc.__cause__ 

187 pfx = 'caused by' 

188 elif exc.__context__: 

189 exc = exc.__context__ 

190 pfx = 'during handling of' 

191 else: 

192 break 

193 

194 if messages: 

195 head += ': ' + messages[0] 

196 if len(lines) > 1: 

197 head += ' ' + lines[1].strip() 

198 

199 lines.insert(0, head) 

200 return lines 

201 

202 

203## 

204 

205 

206def _name(exc): 

207 typ = type(exc) or Exception 

208 # if typ == Error: 

209 # return 'Error' 

210 name = getattr(typ, '__name__', '') 

211 mod = getattr(typ, '__module__', '') 

212 if mod in {'exceptions', 'builtins'}: 

213 return name 

214 return mod + '.' + name 

215 

216 

217def _message(exc): 

218 try: 

219 return repr(exc.args[0]) 

220 except: 

221 return '' 

222 

223 

224## 

225 

226 

227_current_level = Level.INFO 

228 

229_out_stream = sys.stdout 

230 

231_PREFIX = { 

232 Level.CRITICAL: 'CRITICAL', 

233 Level.ERROR: 'ERROR', 

234 Level.WARNING: 'WARNING', 

235 Level.INFO: 'INFO', 

236 Level.DEBUG: 'DEBUG', 

237} 

238 

239_MAX_MSG_LENGTH = 4096 

240 

241 

242def _raw(level, msg, args=None, kwargs=None): 

243 if level < _current_level: 

244 return 

245 

246 if args: 

247 if len(args) == 1 and args[0] and isinstance(args[0], dict): 

248 args = args[0] 

249 msg = msg % args 

250 

251 if len(msg) > _MAX_MSG_LENGTH: 

252 msg = msg[:_MAX_MSG_LENGTH] + '...' 

253 

254 pid = os.getpid() 

255 loc = ' ' 

256 if _current_level <= Level.DEBUG: 

257 stacklevel = kwargs.get('stacklevel', 1) if kwargs else 1 

258 loc = ' ' + _location(2 + stacklevel) + ' ' 

259 pfx = '[' + str(pid) + ']' + loc + _PREFIX[level] + ' :: ' 

260 

261 try: 

262 _out_stream.write(f'{pfx}{msg}\n') 

263 except UnicodeEncodeError: 

264 _out_stream.write(f'{pfx}{msg!r}\n') 

265 

266 _out_stream.flush() 

267 

268 

269def _location(stacklevel): 

270 frames = traceback.extract_stack() 

271 for fname, line, func, text in reversed(frames): 

272 if stacklevel == 0: 

273 return f'{fname}:{line}' 

274 stacklevel -= 1 

275 return '???'