Coverage for functions \ flipdare \ job \ event_data.py: 92%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13""" 

14Event data wrappers for Firestore document changes. 

15 

16Provides simple access to document data and change tracking without the complexity 

17of the full TriggerData validation system. 

18""" 

19 

20from typing import Any, override 

21 

22from flipdare.app_types import DatabaseDict 

23from flipdare.util.debug_util import stringify_debug 

24from flipdare.wrapper import PersistedWrapper 

25 

26__all__ = ["EventData", "UpdateEventData"] 

27 

28 

29class EventData[W: PersistedWrapper[Any]]: 

30 """ 

31 Simple wrapper for Firestore document data. 

32 

33 Provides validated access to document data and model creation. 

34 """ 

35 

36 __slots__ = ("_data", "_doc_id", "_wrapper_class") 

37 

38 def __init__(self, doc_id: str, data: DatabaseDict, wrapper_class: type[W]) -> None: 

39 self._doc_id = doc_id 

40 self._data = data 

41 self._wrapper_class = wrapper_class 

42 

43 @property 

44 def doc_id(self) -> str: 

45 """Document ID.""" 

46 return self._doc_id 

47 

48 @property 

49 def data(self) -> DatabaseDict: 

50 """Raw document data dictionary.""" 

51 return self._data 

52 

53 @property 

54 def wrapper(self) -> PersistedWrapper[Any]: 

55 """Create and return PersistedWrapper from document data.""" 

56 return self._wrapper_class.from_dict(self._data) 

57 

58 def valid(self) -> str | None: 

59 """ 

60 Validate that data can be converted to model. 

61 

62 Returns: 

63 None if valid, error message string if invalid 

64 

65 """ 

66 try: 

67 _ = self._wrapper_class.from_dict(self._data) 

68 return None 

69 except Exception as e: 

70 return str(e) 

71 

72 

73class UpdateEventData[W: PersistedWrapper[Any]](EventData[W]): 

74 """ 

75 Wrapper for Firestore document update events. 

76 

77 Provides access to both before and after states, plus change tracking. 

78 """ 

79 

80 __slots__ = ("_before", "_changes") 

81 

82 def __init__( 

83 self, 

84 doc_id: str, 

85 before: DatabaseDict, 

86 after: DatabaseDict, 

87 wrapper_class: type[W], 

88 ) -> None: 

89 super().__init__(doc_id=doc_id, data=after, wrapper_class=wrapper_class) 

90 self._before = before 

91 self._changes = self._get_changes() 

92 

93 @property 

94 def has_changed(self) -> bool: 

95 """Check if any fields changed between before and after.""" 

96 return self._changes is not None and len(self._changes) > 0 

97 

98 @property 

99 def before(self) -> PersistedWrapper[Any]: 

100 """Create and return PersistedWrapper from before data.""" 

101 return self._wrapper_class.from_dict(self._before) 

102 

103 @property 

104 def before_data(self) -> DatabaseDict: 

105 """Raw before state data dictionary.""" 

106 return self._before 

107 

108 @property 

109 def changes(self) -> DatabaseDict | None: 

110 """Dictionary of changed fields (None if no changes).""" 

111 return self._changes 

112 

113 @override 

114 def valid(self) -> str | None: 

115 """ 

116 Validate that both before and after data can be converted to models. 

117 

118 Returns: 

119 None if valid, error message string if invalid 

120 

121 """ 

122 # Check after data (parent validation) 

123 parent_valid = super().valid() 

124 if parent_valid is not None: 

125 return parent_valid 

126 

127 # Check before data 

128 try: 

129 _ = self._wrapper_class.from_dict(self._before) 

130 return None 

131 except Exception as e: 

132 return str(e) 

133 

134 def _get_changes(self) -> DatabaseDict | None: 

135 """Compute dictionary of changed fields.""" 

136 changes: DatabaseDict = {} 

137 for key in self._data: 

138 before_value = self._before.get(key) 

139 after_value = self._data.get(key) 

140 if before_value != after_value: 

141 changes[key] = after_value 

142 

143 if len(changes) == 0: 

144 return None 

145 

146 return changes 

147 

148 @override 

149 def __repr__(self) -> str: 

150 change_str = stringify_debug(self.changes) if self.changes is not None else "No Changes" 

151 return ( 

152 f"UpdateEventData(doc_id={self.doc_id}, has_changed={self.has_changed})" 

153 "\n" + "-" * 60 + "\n" 

154 f"\nDATA:\n{stringify_debug(self.data)}\n" 

155 "\n" + "-" * 60 + "\n" 

156 f"\nBEFORE:\n{stringify_debug(self.before_data)}\n" 

157 "\n" + "-" * 60 + "\n" 

158 f"\nCHANGES:\n{change_str}\n" 

159 "\n" + "-" * 60 + "\n" 

160 ) 

161 

162 @override 

163 def __str__(self) -> str: 

164 return self.__repr__()