Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/event_bus_service/event_types.py: 100%

1import uuid 

2from datetime import UTC, date, datetime 

3from enum import Enum, auto 

4from typing import Any 

5 

6from pydantic import UUID4, SerializeAsAny, field_validator 

7 

8from ...schema._mealie.mealie_model import MealieModel 

9 

10INTERNAL_INTEGRATION_ID = "mealie_generic_user" 

11 

12 

13class EventTypes(Enum): 

14 """ 

15 The event type defines whether or not a subscriber should receive an event. 

16 

17 Each event type is represented by a field on the subscriber repository, therefore any changes 

18 made here must also be reflected in the database (and likely requires a database migration). 

19 

20 If you'd like more granular control over the metadata of the event, e.g. events for sub-records 

21 (like shopping list items), modify the event document type instead (which is not tied to a database entry). 

22 """ 

23 

24 # used internally and cannot be subscribed to 

25 test_message = auto() 

26 webhook_task = auto() 

27 

28 recipe_created = auto() 

29 recipe_updated = auto() 

30 recipe_deleted = auto() 

31 

32 user_signup = auto() 

33 

34 data_migrations = auto() 

35 data_export = auto() 

36 data_import = auto() 

37 

38 mealplan_entry_created = auto() 

39 

40 shopping_list_created = auto() 

41 shopping_list_updated = auto() 

42 shopping_list_deleted = auto() 

43 

44 cookbook_created = auto() 

45 cookbook_updated = auto() 

46 cookbook_deleted = auto() 

47 

48 tag_created = auto() 

49 tag_updated = auto() 

50 tag_deleted = auto() 

51 

52 category_created = auto() 

53 category_updated = auto() 

54 category_deleted = auto() 

55 

56 label_created = auto() 

57 label_updated = auto() 

58 label_deleted = auto() 

59 

60 

61class EventDocumentType(Enum): 

62 generic = "generic" 

63 

64 user = "user" 

65 

66 category = "category" 

67 cookbook = "cookbook" 

68 mealplan = "mealplan" 

69 shopping_list = "shopping_list" 

70 shopping_list_item = "shopping_list_item" 

71 recipe = "recipe" 

72 recipe_bulk_report = "recipe_bulk_report" 

73 recipe_timeline_event = "recipe_timeline_event" 

74 tag = "tag" 

75 label = "label" 

76 

77 

78class EventOperation(Enum): 

79 info = "info" 

80 

81 create = "create" 

82 update = "update" 

83 delete = "delete" 

84 

85 

86class EventDocumentDataBase(MealieModel): 

87 document_type: EventDocumentType 

88 operation: EventOperation 

89 ... 

90 

91 

92class EventMealplanCreatedData(EventDocumentDataBase): 

93 document_type: EventDocumentType = EventDocumentType.mealplan 

94 operation: EventOperation = EventOperation.create 

95 mealplan_id: int 

96 date: date 

97 recipe_id: UUID4 | None = None 

98 recipe_name: str | None = None 

99 recipe_slug: str | None = None 

100 

101 

102class EventUserSignupData(EventDocumentDataBase): 

103 document_type: EventDocumentType = EventDocumentType.user 

104 operation: EventOperation = EventOperation.create 

105 username: str 

106 email: str 

107 

108 

109class EventCategoryData(EventDocumentDataBase): 

110 document_type: EventDocumentType = EventDocumentType.category 

111 category_id: UUID4 

112 

113 

114class EventLabelData(EventDocumentDataBase): 

115 document_type: EventDocumentType = EventDocumentType.label 

116 label_id: UUID4 

117 

118 

119class EventCookbookData(EventDocumentDataBase): 

120 document_type: EventDocumentType = EventDocumentType.cookbook 

121 cookbook_id: UUID4 

122 

123 

124class EventCookbookBulkData(EventDocumentDataBase): 

125 document_type: EventDocumentType = EventDocumentType.cookbook 

126 cookbook_ids: list[UUID4] 

127 

128 

129class EventShoppingListData(EventDocumentDataBase): 

130 document_type: EventDocumentType = EventDocumentType.shopping_list 

131 shopping_list_id: UUID4 

132 

133 

134class EventShoppingListItemData(EventDocumentDataBase): 

135 document_type: EventDocumentType = EventDocumentType.shopping_list_item 

136 shopping_list_id: UUID4 

137 shopping_list_item_id: UUID4 

138 

139 

140class EventShoppingListItemBulkData(EventDocumentDataBase): 

141 document_type: EventDocumentType = EventDocumentType.shopping_list_item 

142 shopping_list_id: UUID4 

143 shopping_list_item_ids: list[UUID4] 

144 

145 

146class EventRecipeData(EventDocumentDataBase): 

147 document_type: EventDocumentType = EventDocumentType.recipe 

148 recipe_slug: str 

149 

150 

151class EventRecipeBulkData(EventDocumentDataBase): 

152 document_type: EventDocumentType = EventDocumentType.recipe 

153 recipe_slugs: list[str] 

154 

155 

156class EventRecipeBulkReportData(EventDocumentDataBase): 

157 document_type: EventDocumentType = EventDocumentType.recipe_bulk_report 

158 report_id: UUID4 

159 

160 

161class EventRecipeTimelineEventData(EventDocumentDataBase): 

162 document_type: EventDocumentType = EventDocumentType.recipe_timeline_event 

163 recipe_slug: str 

164 recipe_timeline_event_id: UUID4 

165 

166 

167class EventTagData(EventDocumentDataBase): 

168 document_type: EventDocumentType = EventDocumentType.tag 

169 tag_id: UUID4 

170 

171 

172class EventWebhookData(EventDocumentDataBase): 

173 webhook_start_dt: datetime 

174 webhook_end_dt: datetime 

175 webhook_body: Any = None 

176 

177 

178class EventBusMessage(MealieModel): 

179 title: str 

180 body: str = "" 

181 

182 @classmethod 

183 def from_type(cls, event_type: EventTypes, body: str = "") -> "EventBusMessage": 

184 title = event_type.name.replace("_", " ").title() 

185 return cls(title=title, body=body) 

186 

187 @field_validator("body") 

188 def populate_body(v): 

189 # if the body is empty, apprise won't send the notification 

190 return v or "generic" 

191 

192 

193class Event(MealieModel): 

194 message: EventBusMessage 

195 event_type: EventTypes 

196 integration_id: str 

197 document_data: SerializeAsAny[EventDocumentDataBase] 

198 

199 # set at instantiation 

200 event_id: UUID4 | None = None 

201 timestamp: datetime | None = None 

202 

203 def __init__(self, *args, **kwargs) -> None: 

204 super().__init__(*args, **kwargs) 

205 self.event_id = uuid.uuid4() 

206 self.timestamp = datetime.now(UTC)