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

53 statements  

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

1from functools import cached_property 1a

2 

3from fastapi import APIRouter, Depends 1a

4from pydantic import UUID4 1a

5 

6from mealie.routes._base.base_controllers import BaseUserController 1a

7from mealie.routes._base.controller import controller 1a

8from mealie.routes._base.mixins import HttpRepo 1a

9from mealie.routes._base.routers import MealieCrudRoute 1a

10from mealie.schema.household.group_events import ( 1a

11 GroupEventNotifierCreate, 

12 GroupEventNotifierOut, 

13 GroupEventNotifierPrivate, 

14 GroupEventNotifierSave, 

15 GroupEventNotifierUpdate, 

16 GroupEventPagination, 

17) 

18from mealie.schema.mapper import cast 1a

19from mealie.schema.response.pagination import PaginationQuery 1a

20from mealie.services.event_bus_service.event_bus_listeners import AppriseEventListener 1a

21from mealie.services.event_bus_service.event_bus_service import EventBusService 1a

22from mealie.services.event_bus_service.event_types import ( 1a

23 Event, 

24 EventBusMessage, 

25 EventDocumentDataBase, 

26 EventDocumentType, 

27 EventOperation, 

28 EventTypes, 

29) 

30 

31router = APIRouter( 1a

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

33) 

34 

35 

36@controller(router) 1a

37class GroupEventsNotifierController(BaseUserController): 1a

38 event_bus: EventBusService = Depends(EventBusService.as_dependency) 1a

39 

40 @cached_property 1a

41 def repo(self): 1a

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

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

44 

45 return self.repos.group_event_notifier 1dnefghicjkolmb

46 

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

48 # CRUD Operations 

49 

50 @property 1a

51 def mixins(self) -> HttpRepo: 1a

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

53 

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

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

56 response = self.repo.page_all( 1dnefghicjklmb

57 pagination=q, 

58 override=GroupEventNotifierOut, 

59 ) 

60 

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

62 return response 1defghicjklmb

63 

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

65 def create_one(self, data: GroupEventNotifierCreate): 1a

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

67 return self.mixins.create_one(save_data) 1dnefghicjkolmb

68 

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

70 def get_one(self, item_id: UUID4): 1a

71 return self.mixins.get_one(item_id) 

72 

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

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

75 if data.apprise_url is None: 1cb

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) 1cb

82 

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

84 def delete_one(self, item_id: UUID4): 1a

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

92 def test_notification(self, item_id: UUID4): 1a

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])