Coverage for /opt/conda/envs/apienv/lib/python3.10/site-packages/daiquiri/core/components/logging.py: 75%

1#!/usr/bin/env python 

2# -*- coding: utf-8 -*- 

3import json 

4import logging 

5 

6from marshmallow import Schema, fields 

7from flask import g 

8 

9from daiquiri.core import marshal 

10from daiquiri.core.logging import log 

11from daiquiri.core.components import Component, ComponentResource 

12from daiquiri.core.schema import MessageSchema, ErrorSchema 

13from daiquiri.core.exceptions import PrettyException 

14from daiquiri.core.schema.validators import OneOf 

15 

16logger = logging.getLogger(__name__) 

17 

18 

19class UIError(PrettyException): 

20 """A UI Error Exception (pretty)""" 

21 

22 def __init__(self, message, frame=None, error=None): 

23 super().__init__(message) 

24 self.frame = frame 

25 self.error = error 

26 self.message = message 

27 

28 def pretty(self): 

29 message = "A UI Error Occured\n" 

30 message += f" {self.message}\n" 

31 if self.frame: 

32 message += f" at {self.frame}\n" 

33 else: 

34 message += " at [Could not resolve stacktrace]\n" 

35 message += f" {self.error}\n" 

36 

37 return message 

38 

39 

40class UIErrorSchema(Schema): 

41 status = OneOf( 

42 ["resolved", "error"], 

43 required=True, 

44 metadata={"description": "Whether the frame stacktrace was resolved"}, 

45 ) 

46 message = fields.Str(required=True, metadata={"description": "The error messsage"}) 

47 frame = fields.Str(metadata={"description": "The frame stacktrace"}) 

48 error = fields.Str( 

49 metadata={ 

50 "description": "The (secondary) error as to why the frame couldnt be resolved" 

51 } 

52 ) 

53 

54 

55class LogSchema(Schema): 

56 logs = fields.Dict() 

57 

58 

59class LogResource(ComponentResource): 

60 @marshal( 

61 inp={"lines": fields.Int(), "offset": fields.Int()}, 

62 out=[ 

63 [200, LogSchema(), "Lines of log file"], 

64 [400, ErrorSchema(), "No access to log file"], 

65 ], 

66 ) 

67 def get(self, **kwargs): 

68 """Get the last n lines of the user level log""" 

69 if g.user.staff(): 

70 return {"logs": self._parent.get_logs(**kwargs)} 

71 else: 

72 return {"error": "No access"}, 400 

73 

74 

75class UILogResource(ComponentResource): 

76 @marshal( 

77 inp=UIErrorSchema, 

78 out=[ 

79 [200, MessageSchema(), "Error logged successfully"], 

80 [400, ErrorSchema(), "Could not log error"], 

81 ], 

82 ) 

83 def post(self, **kwargs): 

84 """Log a UI Error""" 

85 message = kwargs.pop("message") 

86 try: 

87 raise UIError(message, frame=kwargs.get("frame"), error=kwargs.get("error")) 

88 except UIError as e: 

89 log.get("ui").exception(message, type="app", state=kwargs.get("store")) 

90 logger.error(e.pretty()) 

91 return {"message": "Error logged"} 

92 

93 

94class Logging(Component): 

95 """Logging Component 

96 

97 Allows the application to retrieve historic logs from the json log file 

98 (to allow debuggin) 

99 """ 

100 

101 def setup(self, *args, **kwargs): 

102 self.register_route(LogResource, "") 

103 self.register_route(UILogResource, "/ui") 

104 

105 def get_logs(self, **kwargs): 

106 lines = kwargs.get("lines", 50) 

107 offset = kwargs.get("offset", 0) 

108 

109 out = {} 

110 for ty, file in log.list_files().items(): 

111 if ty == "ui": 

112 continue 

113 

114 with open(file, "rb") as f: 

115 out[ty] = [ 

116 json.loads(line.decode("utf-8")) 

117 for line in self.tail(f, lines + offset)[::-1][offset:] 

118 ] 

119 

120 return out 

121 

122 def tail(self, f, window=1): 

123 """ 

124 Returns the last `window` lines of file `f` as a list of bytes. 

125 https://stackoverflow.com/questions/136168/get-last-n-lines-of-a-file-with-python-similar-to-tail 

126 """ 

127 if window == 0: 

128 return b"" 

129 BUFSIZE = 1024 

130 f.seek(0, 2) 

131 end = f.tell() 

132 nlines = window + 1 

133 data = [] 

134 while nlines > 0 and end > 0: 

135 i = max(0, end - BUFSIZE) 

136 nread = min(end, BUFSIZE) 

137 

138 f.seek(i) 

139 chunk = f.read(nread) 

140 data.append(chunk) 

141 nlines -= chunk.count(b"\n") 

142 end -= nread 

143 return b"".join(reversed(data)).splitlines()[-window:]