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

1from typing import Protocol 

2 

3import apprise 

4import requests 

5from fastapi.encoders import jsonable_encoder 

6 

7from mealie.services.event_bus_service.event_types import Event 

8 

9 

10class PublisherLike(Protocol): 

11 def publish(self, event: Event, notification_urls: list[str]): ... 11 ↛ exitline 11 didn't return from function 'publish' because

12 

13 

14class ApprisePublisher: 

15 def __init__(self, hard_fail=False) -> None: 

16 asset = apprise.AppriseAsset( 

17 async_mode=True, 

18 image_url_mask="https://raw.githubusercontent.com/mealie-recipes/mealie/9571816ac4eed5beacfc0abf6c03eff1427fd0eb/frontend/static/icons/android-chrome-maskable-512x512.png", 

19 ) 

20 self.apprise = apprise.Apprise(asset=asset) 

21 self.hard_fail = hard_fail 

22 

23 def publish(self, event: Event, notification_urls: list[str]): 

24 """Publishses a list of notification URLs""" 

25 

26 tags = [] 

27 for dest in notification_urls: 

28 # we tag the url so it only sends each notification once 

29 tag = str(event.event_id) 

30 tags.append(tag) 

31 

32 status = self.apprise.add(dest, tag=tag) 

33 

34 if not status and self.hard_fail: 

35 raise Exception("Apprise URL Add Failed") 

36 

37 self.apprise.notify(title=event.message.title, body=event.message.body, tag=tags) 

38 

39 

40class WebhookPublisher: 

41 def __init__(self, hard_fail=False) -> None: 

42 self.hard_fail = hard_fail 

43 

44 def publish(self, event: Event, notification_urls: list[str]): 

45 event_payload = jsonable_encoder(event) 

46 for url in notification_urls: 46 ↛ exitline 46 didn't return from function 'publish' because the loop on line 46 didn't complete

47 r = requests.post(url, json=event_payload, timeout=15) 

48 if self.hard_fail: 

49 r.raise_for_status()