Coverage for functions \ flipdare \ error \ log_context.py: 92%

1#!/usr/bin/env python 

2from __future__ import annotations 

3 

4from datetime import datetime, UTC 

5from typing import TYPE_CHECKING, Any 

6from dataclasses import dataclass, field 

7from flipdare.app_types import DatabaseDict 

8from flipdare.constants import EMAIL_LOG_SUBJECT_MAX_LENGTH, NO_DOC_ID 

9from flipdare.generated.shared.app_log_category import AppLogCategory 

10from flipdare.result.app_result import AppResult 

11from flipdare.error.app_error_protocol import AppErrorProtocol 

12from flipdare.generated.model.backend.app_log_model import AppLogModel 

13from flipdare.generated.schema.email.body.admin.log_email_schema import LogEmailSchema 

14from flipdare.generated.shared.app_error_code import AppErrorCode 

15from flipdare.generated.shared.backend.app_job_type import AppJobType 

16from flipdare.generated.shared.backend.system_log_type import SystemLogType 

17from flipdare.generated.shared.firestore_collections import FirestoreCollections 

18from flipdare.generated.shared.search.search_collections import SearchCollections 

19 

20if TYPE_CHECKING: 

21 from flipdare.error.message_format import BaseMsgFormat 

22 

23 

24__all__ = ["LogContext"] 

25 

26 

27@dataclass(frozen=True, kw_only=True) 

28class LogContext: 

29 # core 

30 log_type: SystemLogType 

31 called_by: str 

32 message: str 

33 doc_id: str = NO_DOC_ID 

34 

35 category: AppLogCategory 

36 

37 # optional, since we could have info. 

38 error_code: AppErrorProtocol | None = None 

39 

40 source_override: str | None = ( 

41 None # this is used to construct the source (fallback is called_by) 

42 ) 

43 # this is used to construct the source (fallback is called_by) 

44 collection: FirestoreCollections | SearchCollections | None = None 

45 job_type: AppJobType | None = None 

46 

47 # misc 

48 duration: int | None = None 

49 notify_admin: bool = False 

50 result: AppResult[Any] | None = None 

51 

52 # data 

53 stack_trace: str | None = None 

54 formatter: BaseMsgFormat | None = None 

55 

56 # auto generated 

57 _occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

58 _extra: DatabaseDict | str | None = None 

59 

60 @property 

61 def occurred_at(self) -> str: 

62 return self._occurred_at.strftime("%Y-%m-%d %H:%M:%S UTC") 

63 

64 @property 

65 def error_code_str(self) -> str: 

66 return str(self.error_code) if self.error_code else AppErrorCode.SERVER.value 

67 

68 @property 

69 def source(self) -> str: 

70 return ( 

71 self.source_override 

72 or self.job_type 

73 or self.collection 

74 or self.called_by 

75 or self.category 

76 ) 

77 

78 @property 

79 def subject(self) -> str: 

80 msg = f"[{self.log_type.value}]: " 

81 msg += f"{self.source} - " 

82 msg += ( 

83 f"{self.message[:EMAIL_LOG_SUBJECT_MAX_LENGTH]}..." 

84 if len(self.message) > EMAIL_LOG_SUBJECT_MAX_LENGTH 

85 else self.message 

86 ) 

87 return msg 

88 

89 @property 

90 def extra(self) -> DatabaseDict: 

91 actual_extra: dict[str, Any] = {} 

92 extra = self._extra 

93 if extra is not None: 

94 actual_extra["extra"] = extra 

95 

96 if result := self.result: 

97 actual_extra["result"] = result.to_dict() 

98 

99 return actual_extra 

100 

101 @property 

102 def email(self) -> LogEmailSchema: 

103 schema = LogEmailSchema( 

104 occurred_at=self.occurred_at, 

105 log_type=self.log_type, 

106 error_code=self.error_code_str, 

107 called_from=self.called_by, 

108 source=self.source, 

109 message=self.message, 

110 detail=self.formatted, 

111 ) 

112 

113 if extra := self._extra: 

114 if isinstance(extra, str): 

115 schema["extra"] = {"info": extra} 

116 else: 

117 schema["extra"] = extra 

118 

119 if stack := self.stack_trace: 

120 schema["stack_trace"] = stack 

121 

122 return schema 

123 

124 @property 

125 def log_model(self) -> AppLogModel: 

126 log_type = self.log_type 

127 acknowledged = False 

128 if log_type in (SystemLogType.WARNING, SystemLogType.INFO): 

129 acknowledged = True 

130 

131 firestore_collection: FirestoreCollections | None = None 

132 search_collection: SearchCollections | None = None 

133 if isinstance(self.collection, FirestoreCollections): 

134 firestore_collection = self.collection 

135 elif isinstance(self.collection, SearchCollections): 

136 search_collection = self.collection 

137 

138 return AppLogModel( 

139 id=None, 

140 category=self.category, 

141 log_type=log_type, 

142 source=self.source, 

143 firestore_collection=firestore_collection, 

144 search_collection=search_collection, 

145 called_by=self.called_by, 

146 admin_notified=self.notify_admin, 

147 acknowledged=acknowledged, 

148 message=self.message, 

149 job_type=self.job_type or None, 

150 error_code=self.error_code or None, 

151 obj_id=self.doc_id, 

152 extra=self.extra, 

153 stack_trace=self.stack_trace, 

154 ) 

155 

156 @property 

157 def formatted(self) -> str: 

158 from flipdare.error.message_format import ErrorMsgFormat, InfoMsgFormat 

159 

160 if self.formatter is not None: 

161 return str(self.formatter) 

162 

163 if self.log_type == SystemLogType.INFO: 

164 return str(InfoMsgFormat(self)) 

165 

166 return str(ErrorMsgFormat(self))