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

132 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 17:29 +0000

1import uuid 1a

2from datetime import UTC, date, datetime 1a

3from enum import Enum, auto 1a

4from typing import Any 1a

5 

6from pydantic import UUID4, SerializeAsAny, field_validator 1a

7 

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

9 

10INTERNAL_INTEGRATION_ID = "mealie_generic_user" 1a

11 

12 

13class EventTypes(Enum): 1a

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() 1a

26 webhook_task = auto() 1a

27 

28 recipe_created = auto() 1a

29 recipe_updated = auto() 1a

30 recipe_deleted = auto() 1a

31 

32 user_signup = auto() 1a

33 

34 data_migrations = auto() 1a

35 data_export = auto() 1a

36 data_import = auto() 1a

37 

38 mealplan_entry_created = auto() 1a

39 

40 shopping_list_created = auto() 1a

41 shopping_list_updated = auto() 1a

42 shopping_list_deleted = auto() 1a

43 

44 cookbook_created = auto() 1a

45 cookbook_updated = auto() 1a

46 cookbook_deleted = auto() 1a

47 

48 tag_created = auto() 1a

49 tag_updated = auto() 1a

50 tag_deleted = auto() 1a

51 

52 category_created = auto() 1a

53 category_updated = auto() 1a

54 category_deleted = auto() 1a

55 

56 label_created = auto() 1a

57 label_updated = auto() 1a

58 label_deleted = auto() 1a

59 

60 

61class EventDocumentType(Enum): 1a

62 generic = "generic" 1a

63 

64 user = "user" 1a

65 

66 category = "category" 1a

67 cookbook = "cookbook" 1a

68 mealplan = "mealplan" 1a

69 shopping_list = "shopping_list" 1a

70 shopping_list_item = "shopping_list_item" 1a

71 recipe = "recipe" 1a

72 recipe_bulk_report = "recipe_bulk_report" 1a

73 recipe_timeline_event = "recipe_timeline_event" 1a

74 tag = "tag" 1a

75 label = "label" 1a

76 

77 

78class EventOperation(Enum): 1a

79 info = "info" 1a

80 

81 create = "create" 1a

82 update = "update" 1a

83 delete = "delete" 1a

84 

85 

86class EventDocumentDataBase(MealieModel): 1a

87 document_type: EventDocumentType 1a

88 operation: EventOperation 1a

89 ... 1a

90 

91 

92class EventMealplanCreatedData(EventDocumentDataBase): 1a

93 document_type: EventDocumentType = EventDocumentType.mealplan 1a

94 operation: EventOperation = EventOperation.create 1a

95 mealplan_id: int 1a

96 date: date 1a

97 recipe_id: UUID4 | None = None 1a

98 recipe_name: str | None = None 1a

99 recipe_slug: str | None = None 1a

100 

101 

102class EventUserSignupData(EventDocumentDataBase): 1a

103 document_type: EventDocumentType = EventDocumentType.user 1a

104 operation: EventOperation = EventOperation.create 1a

105 username: str 1a

106 email: str 1a

107 

108 

109class EventCategoryData(EventDocumentDataBase): 1a

110 document_type: EventDocumentType = EventDocumentType.category 1a

111 category_id: UUID4 1a

112 

113 

114class EventLabelData(EventDocumentDataBase): 1a

115 document_type: EventDocumentType = EventDocumentType.label 1a

116 label_id: UUID4 1a

117 

118 

119class EventCookbookData(EventDocumentDataBase): 1a

120 document_type: EventDocumentType = EventDocumentType.cookbook 1a

121 cookbook_id: UUID4 1a

122 

123 

124class EventCookbookBulkData(EventDocumentDataBase): 1a

125 document_type: EventDocumentType = EventDocumentType.cookbook 1a

126 cookbook_ids: list[UUID4] 1a

127 

128 

129class EventShoppingListData(EventDocumentDataBase): 1a

130 document_type: EventDocumentType = EventDocumentType.shopping_list 1a

131 shopping_list_id: UUID4 1a

132 

133 

134class EventShoppingListItemData(EventDocumentDataBase): 1a

135 document_type: EventDocumentType = EventDocumentType.shopping_list_item 1a

136 shopping_list_id: UUID4 1a

137 shopping_list_item_id: UUID4 1a

138 

139 

140class EventShoppingListItemBulkData(EventDocumentDataBase): 1a

141 document_type: EventDocumentType = EventDocumentType.shopping_list_item 1a

142 shopping_list_id: UUID4 1a

143 shopping_list_item_ids: list[UUID4] 1a

144 

145 

146class EventRecipeData(EventDocumentDataBase): 1a

147 document_type: EventDocumentType = EventDocumentType.recipe 1a

148 recipe_slug: str 1a

149 

150 

151class EventRecipeBulkData(EventDocumentDataBase): 1a

152 document_type: EventDocumentType = EventDocumentType.recipe 1a

153 recipe_slugs: list[str] 1a

154 

155 

156class EventRecipeBulkReportData(EventDocumentDataBase): 1a

157 document_type: EventDocumentType = EventDocumentType.recipe_bulk_report 1a

158 report_id: UUID4 1a

159 

160 

161class EventRecipeTimelineEventData(EventDocumentDataBase): 1a

162 document_type: EventDocumentType = EventDocumentType.recipe_timeline_event 1a

163 recipe_slug: str 1a

164 recipe_timeline_event_id: UUID4 1a

165 

166 

167class EventTagData(EventDocumentDataBase): 1a

168 document_type: EventDocumentType = EventDocumentType.tag 1a

169 tag_id: UUID4 1a

170 

171 

172class EventWebhookData(EventDocumentDataBase): 1a

173 webhook_start_dt: datetime 1a

174 webhook_end_dt: datetime 1a

175 webhook_body: Any = None 1a

176 

177 

178class EventBusMessage(MealieModel): 1a

179 title: str 1a

180 body: str = "" 1a

181 

182 @classmethod 1a

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

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

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

186 

187 @field_validator("body") 1a

188 def populate_body(v): 1a

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

190 return v or "generic" 1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ

191 

192 

193class Event(MealieModel): 1a

194 message: EventBusMessage 1a

195 event_type: EventTypes 1a

196 integration_id: str 1a

197 document_data: SerializeAsAny[EventDocumentDataBase] 1a

198 

199 # set at instantiation 

200 event_id: UUID4 | None = None 1a

201 timestamp: datetime | None = None 1a

202 

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

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

205 self.event_id = uuid.uuid4() 1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ

206 self.timestamp = datetime.now(UTC) 1bcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ