Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/households/controller_group_notifications.py: 96%

1from functools import cached_property 

2 

3from fastapi import APIRouter, Depends 

4from pydantic import UUID4 

5 

6from mealie.routes._base.base_controllers import BaseUserController 

7from mealie.routes._base.controller import controller 

8from mealie.routes._base.mixins import HttpRepo 

9from mealie.routes._base.routers import MealieCrudRoute 

10from mealie.schema.household.group_events import ( 

11 GroupEventNotifierCreate, 

12 GroupEventNotifierOut, 

13 GroupEventNotifierPrivate, 

14 GroupEventNotifierSave, 

15 GroupEventNotifierUpdate, 

16 GroupEventPagination, 

17) 

18from mealie.schema.mapper import cast 

19from mealie.schema.response.pagination import PaginationQuery 

20from mealie.services.event_bus_service.event_bus_listeners import AppriseEventListener 

21from mealie.services.event_bus_service.event_bus_service import EventBusService 

22from mealie.services.event_bus_service.event_types import ( 

23 Event, 

24 EventBusMessage, 

25 EventDocumentDataBase, 

26 EventDocumentType, 

27 EventOperation, 

28 EventTypes, 

29) 

30 

31router = APIRouter( 

32 prefix="/households/events/notifications", tags=["Households: Event Notifications"], route_class=MealieCrudRoute 

33) 

34 

35 

36@controller(router) 

37class GroupEventsNotifierController(BaseUserController): 

38 event_bus: EventBusService = Depends(EventBusService.as_dependency) 

39 

40 @cached_property 

41 def repo(self): 

42 if not self.user: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true

43 raise Exception("No user is logged in.") 

44 

45 return self.repos.group_event_notifier 

46 

47 # ======================================================================= 

48 # CRUD Operations 

49 

50 @property 

51 def mixins(self) -> HttpRepo: 

52 return HttpRepo(self.repo, self.logger, self.registered_exceptions, self.t("generic.server-error")) 

53 

54 @router.get("", response_model=GroupEventPagination) 

55 def get_all(self, q: PaginationQuery = Depends(PaginationQuery)): 

56 response = self.repo.page_all( 

57 pagination=q, 

58 override=GroupEventNotifierOut, 

59 ) 

60 

61 response.set_pagination_guides(router.url_path_for("get_all"), q.model_dump()) 

62 return response 

63 

64 @router.post("", response_model=GroupEventNotifierOut, status_code=201) 

65 def create_one(self, data: GroupEventNotifierCreate): 

66 save_data = cast(data, GroupEventNotifierSave, group_id=self.group_id, household_id=self.household_id) 

67 return self.mixins.create_one(save_data) 

68 

69 @router.get("/{item_id}", response_model=GroupEventNotifierOut) 

70 def get_one(self, item_id: UUID4): 

71 return self.mixins.get_one(item_id) 

72 

73 @router.put("/{item_id}", response_model=GroupEventNotifierOut) 

74 def update_one(self, item_id: UUID4, data: GroupEventNotifierUpdate): 

75 if data.apprise_url is None: 

76 current_data: GroupEventNotifierPrivate = self.repo.get_one( 

77 item_id, override_schema=GroupEventNotifierPrivate 

78 ) 

79 data.apprise_url = current_data.apprise_url 

80 

81 return self.mixins.update_one(data, item_id) 

82 

83 @router.delete("/{item_id}", status_code=204) 

84 def delete_one(self, item_id: UUID4): 

85 self.mixins.delete_one(item_id) # type: ignore 

86 

87 # ======================================================================= 

88 # Test Event Notifications 

89 

90 # TODO: properly re-implement this with new event listeners 

91 @router.post("/{item_id}/test", status_code=204) 

92 def test_notification(self, item_id: UUID4): 

93 item: GroupEventNotifierPrivate = self.repo.get_one(item_id, override_schema=GroupEventNotifierPrivate) 

94 

95 event_type = EventTypes.test_message 

96 test_event = Event( 

97 message=EventBusMessage.from_type(event_type, "test message"), 

98 event_type=event_type, 

99 integration_id="test_event", 

100 document_data=EventDocumentDataBase(document_type=EventDocumentType.generic, operation=EventOperation.info), 

101 ) 

102 

103 test_listener = AppriseEventListener(self.group_id, self.household_id) 

104 test_listener.publish_to_subscribers(test_event, [item.apprise_url])