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 ls = exception_backtrace(exc) 

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

127 for s in ls[1:]: 

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

129 

130 

131def if_debug(fn, *args): 

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

133 

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

135 

136 Args: 

137 fn: A function that returns a message. 

138 *args: Arguments for the function. 

139 """ 

140 

141 if Level.DEBUG < _current_level: 

142 return 

143 try: 

144 msg = fn(*args) 

145 except Exception as exc: 

146 msg = repr(exc) 

147 _raw(Level.DEBUG, msg) 

148 

149 

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

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

152 

153 The backtrace includes the chain of causes and contexts. 

154 

155 Args: 

156 exc: An exception. 

157 

158 Returns: 

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

160 """ 

161 

162 head = _name(exc) 

163 messages = [] 

164 

165 lines = [] 

166 pfx = '' 

167 

168 while exc: 

169 subhead = _name(exc) 

170 msg = _message(exc) 

171 if msg: 

172 subhead += ': ' + msg 

173 messages.append(msg) 

174 if pfx: 

175 subhead = pfx + ' ' + subhead 

176 

177 lines.append(subhead) 

178 

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

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

181 

182 if exc.__cause__: 

183 exc = exc.__cause__ 

184 pfx = 'caused by' 

185 elif exc.__context__: 

186 exc = exc.__context__ 

187 pfx = 'during handling of' 

188 else: 

189 break 

190 

191 if messages: 

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

193 if len(lines) > 1: 

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

195 

196 lines.insert(0, head) 

197 return lines 

198 

199 

200## 

201 

202 

203def _name(exc): 

204 typ = type(exc) or Exception 

205 # if typ == Error: 

206 # return 'Error' 

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

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

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

210 return name 

211 return mod + '.' + name 

212 

213 

214def _message(exc): 

215 try: 

216 return repr(exc.args[0]) 

217 except: 

218 return '' 

219 

220 

221## 

222 

223 

224_current_level = Level.INFO 

225 

226_out_stream = sys.stdout 

227 

228_PREFIX = { 

229 Level.CRITICAL: 'CRITICAL', 

230 Level.ERROR: 'ERROR', 

231 Level.WARNING: 'WARNING', 

232 Level.INFO: 'INFO', 

233 Level.DEBUG: 'DEBUG', 

234} 

235 

236_MAX_MSG_LENGTH = 4096 

237 

238 

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

240 if level < _current_level: 

241 return 

242 

243 if args: 

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

245 args = args[0] 

246 msg = msg % args 

247 

248 if len(msg) > _MAX_MSG_LENGTH: 

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

250 

251 pid = os.getpid() 

252 loc = ' ' 

253 if _current_level <= Level.DEBUG: 

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

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

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

257 

258 try: 

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

260 except UnicodeEncodeError: 

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

262 

263 _out_stream.flush() 

264 

265 

266def _location(stacklevel): 

267 frames = traceback.extract_stack() 

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

269 if stacklevel == 0: 

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

271 stacklevel -= 1 

272 return '???'