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

1"""Logging facility.""" 

2 

3import os 

4import sys 

5import traceback 

6 

7 

8class Level: 

9 """Log level.""" 

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 global _current_level 

23 if level is None: 

24 _current_level = Level.INFO 

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

26 _current_level = int(level) 

27 else: 

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

29 

30 

31def get_level() -> str: 

32 global _current_level 

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

34 if n == _current_level: 

35 return k 

36 return 'ALL' 

37 

38 

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

40 _raw(level, msg, args, kwargs) 

41 

42 

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

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

45 

46 

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

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

49 

50 

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

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

53 

54 

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

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

57 

58 

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

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

61 

62 

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

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

65 ls = exception_backtrace(exc) 

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

67 for s in ls[1:]: 

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

69 

70 

71def if_debug(fn, *args): 

72 """If debugging, apply the function to args and log the result.""" 

73 

74 if Level.DEBUG < _current_level: 

75 return 

76 try: 

77 msg = fn(*args) 

78 except Exception as exc: 

79 msg = repr(exc) 

80 _raw(Level.DEBUG, msg) 

81 

82 

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

84 """Exception backtrace as a list of strings.""" 

85 

86 head = _name(exc) 

87 messages = [] 

88 

89 lines = [] 

90 pfx = '' 

91 

92 while exc: 

93 subhead = _name(exc) 

94 msg = _message(exc) 

95 if msg: 

96 subhead += ': ' + msg 

97 messages.append(msg) 

98 if pfx: 

99 subhead = pfx + ' ' + subhead 

100 

101 lines.append(subhead) 

102 

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

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

105 

106 if exc.__cause__: 

107 exc = exc.__cause__ 

108 pfx = 'caused by' 

109 elif exc.__context__: 

110 exc = exc.__context__ 

111 pfx = 'during handling of' 

112 else: 

113 break 

114 

115 if messages: 

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

117 if len(lines) > 1: 

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

119 

120 lines.insert(0, head) 

121 return lines 

122 

123 

124## 

125 

126 

127def _name(exc): 

128 typ = type(exc) or Exception 

129 # if typ == Error: 

130 # return 'Error' 

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

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

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

134 return name 

135 return mod + '.' + name 

136 

137 

138def _message(exc): 

139 try: 

140 return repr(exc.args[0]) 

141 except: 

142 return '' 

143 

144 

145## 

146 

147 

148_current_level = Level.INFO 

149 

150_out_stream = sys.stdout 

151 

152_PREFIX = { 

153 Level.CRITICAL: 'CRITICAL', 

154 Level.ERROR: 'ERROR', 

155 Level.WARNING: 'WARNING', 

156 Level.INFO: 'INFO', 

157 Level.DEBUG: 'DEBUG', 

158} 

159 

160_MAX_MSG_LENGTH = 4096 

161 

162 

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

164 if level < _current_level: 

165 return 

166 

167 if args: 

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

169 args = args[0] 

170 msg = msg % args 

171 

172 if len(msg) > _MAX_MSG_LENGTH: 

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

174 

175 pid = os.getpid() 

176 loc = ' ' 

177 if _current_level <= Level.DEBUG: 

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

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

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

181 

182 try: 

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

184 except UnicodeEncodeError: 

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

186 

187 _out_stream.flush() 

188 

189 

190def _location(stacklevel): 

191 frames = traceback.extract_stack() 

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

193 if stacklevel == 0: 

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

195 stacklevel -= 1 

196 return '???'