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

1"""Logging facility.""" 

2 

3import os 

4import sys 

5import traceback 

6 

7 

8class Level: 

9 """Log level.""" 

10 CRITICAL = 50 

11 ERROR = 40 

12 WARN = 30 

13 WARNING = 30 

14 INFO = 20 

15 DEBUG = 10 

16 NOTSET = 0 

17 ALL = 0 

18 

19 

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

21 global _current_level 

22 if level is None: 

23 _current_level = Level.INFO 

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

25 _current_level = int(level) 

26 else: 

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

28 

29 

30def get_level() -> str: 

31 global _current_level 

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

33 if n == _current_level: 

34 return k 

35 return 'ALL' 

36 

37 

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

39 _raw(level, msg, args, kwargs) 

40 

41 

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

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

44 

45 

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

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

48 

49 

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

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

52 

53 

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

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

56 

57 

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

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

60 

61 

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

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

64 ls = exception_backtrace(exc) 

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

66 for s in ls[1:]: 

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

68 

69 

70def if_debug(fn, *args): 

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

72 

73 if Level.DEBUG < _current_level: 

74 return 

75 try: 

76 msg = fn(*args) 

77 except Exception as exc: 

78 msg = repr(exc) 

79 _raw(Level.DEBUG, msg) 

80 

81 

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

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

84 

85 head = _name(exc) 

86 messages = [] 

87 

88 lines = [] 

89 pfx = '' 

90 

91 while exc: 

92 subhead = _name(exc) 

93 msg = _message(exc) 

94 if msg: 

95 subhead += ': ' + msg 

96 messages.append(msg) 

97 if pfx: 

98 subhead = pfx + ' ' + subhead 

99 

100 lines.append(subhead) 

101 

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

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

104 

105 if exc.__cause__: 

106 exc = exc.__cause__ 

107 pfx = 'caused by' 

108 elif exc.__context__: 

109 exc = exc.__context__ 

110 pfx = 'during handling of' 

111 else: 

112 break 

113 

114 if messages: 

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

116 if len(lines) > 1: 

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

118 

119 lines.insert(0, head) 

120 return lines 

121 

122 

123## 

124 

125def _name(exc): 

126 typ = type(exc) or Exception 

127 # if typ == Error: 

128 # return 'Error' 

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

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

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

132 return name 

133 return mod + '.' + name 

134 

135 

136def _message(exc): 

137 try: 

138 return repr(exc.args[0]) 

139 except: 

140 return '' 

141 

142 

143## 

144 

145 

146_current_level = Level.INFO 

147 

148_out_stream = sys.stdout 

149 

150_PREFIX = { 

151 Level.CRITICAL: 'CRITICAL', 

152 Level.ERROR: 'ERROR', 

153 Level.WARNING: 'WARNING', 

154 Level.INFO: 'INFO', 

155 Level.DEBUG: 'DEBUG', 

156} 

157 

158 

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

160 if level < _current_level: 

161 return 

162 

163 if args: 

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

165 args = args[0] 

166 msg = msg % args 

167 

168 pid = os.getpid() 

169 loc = ' ' 

170 if _current_level <= Level.DEBUG: 

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

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

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

174 

175 try: 

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

177 except UnicodeEncodeError: 

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

179 

180 _out_stream.flush() 

181 

182 

183def _location(stacklevel): 

184 frames = traceback.extract_stack() 

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

186 if stacklevel == 0: 

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

188 stacklevel -= 1 

189 return '???'