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

1"""Logging facility.""" 

2 

3import os 

4import sys 

5import traceback 

6 

7 

8class Level: 

9 CRITICAL = 50 

10 ERROR = 40 

11 WARN = 30 

12 WARNING = 30 

13 INFO = 20 

14 DEBUG = 10 

15 NOTSET = 0 

16 ALL = 0 

17 

18 

19def set_level(level: int | str): 

20 global _current_level 

21 if isinstance(level, int) or level.isdigit(): 

22 _current_level = int(level) 

23 else: 

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

25 

26 

27def get_level() -> str: 

28 global _current_level 

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

30 if n == _current_level: 

31 return k 

32 return 'ALL' 

33 

34 

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

36 _raw(level, msg, args, kwargs) 

37 

38 

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

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

41 

42 

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

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

45 

46 

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

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

49 

50 

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

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

53 

54 

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

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

57 

58 

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

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

61 ls = exception_backtrace(exc) 

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

63 for s in ls[1:]: 

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

65 

66 

67def if_debug(fn, *args): 

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

69 

70 if Level.DEBUG < _current_level: 

71 return 

72 try: 

73 msg = fn(*args) 

74 except Exception as exc: 

75 msg = repr(exc) 

76 _raw(Level.DEBUG, msg) 

77 

78 

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

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

81 

82 head = _name(exc) 

83 messages = [] 

84 

85 lines = [] 

86 pfx = '' 

87 

88 while exc: 

89 subhead = _name(exc) 

90 msg = _message(exc) 

91 if msg: 

92 subhead += ': ' + msg 

93 messages.append(msg) 

94 if pfx: 

95 subhead = pfx + ' ' + subhead 

96 

97 lines.append(subhead) 

98 

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

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

101 

102 if exc.__cause__: 

103 exc = exc.__cause__ 

104 pfx = 'caused by' 

105 elif exc.__context__: 

106 exc = exc.__context__ 

107 pfx = 'during handling of' 

108 else: 

109 break 

110 

111 if messages: 

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

113 if len(lines) > 1: 

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

115 

116 lines.insert(0, head) 

117 return lines 

118 

119 

120## 

121 

122def _name(exc): 

123 typ = type(exc) or Exception 

124 # if typ == Error: 

125 # return 'Error' 

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

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

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

129 return name 

130 return mod + '.' + name 

131 

132 

133def _message(exc): 

134 try: 

135 return repr(exc.args[0]) 

136 except: 

137 return '' 

138 

139 

140## 

141 

142 

143_current_level = Level.INFO 

144 

145_out_stream = sys.stdout 

146 

147_PREFIX = { 

148 Level.CRITICAL: 'CRITICAL', 

149 Level.ERROR: 'ERROR', 

150 Level.WARNING: 'WARNING', 

151 Level.INFO: 'INFO', 

152 Level.DEBUG: 'DEBUG', 

153} 

154 

155 

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

157 if level < _current_level: 

158 return 

159 

160 if args: 

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

162 args = args[0] 

163 msg = msg % args 

164 

165 pid = os.getpid() 

166 loc = ' ' 

167 if _current_level <= Level.DEBUG: 

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

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

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

171 

172 try: 

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

174 except UnicodeEncodeError: 

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

176 

177 _out_stream.flush() 

178 

179 

180def _location(stacklevel): 

181 frames = traceback.extract_stack() 

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

183 if stacklevel == 0: 

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

185 stacklevel -= 1 

186 return '???'