Coverage for src/couchers/notifications/push.py: 96%

23 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-20 21:46 +0000

1import functools 

2import json 

3 

4from sqlalchemy.sql import func 

5 

6from couchers import urls 

7from couchers.config import config 

8from couchers.db import session_scope 

9from couchers.jobs.enqueue import queue_job 

10from couchers.models import PushNotificationSubscription 

11from couchers.notifications.push_api import get_vapid_public_key_from_private_key 

12from couchers.sql import couchers_select as select 

13from proto.internal import jobs_pb2 

14 

15 

16@functools.lru_cache 

17def get_vapid_public_key(): 

18 return get_vapid_public_key_from_private_key(config["PUSH_NOTIFICATIONS_VAPID_PRIVATE_KEY"]) 

19 

20 

21def push_to_subscription( 

22 push_notification_subscription_id: int, *, title: str, body: str, icon: str = None, url: str = None, ttl: int = 0 

23): 

24 queue_job( 

25 job_type="send_raw_push_notification", 

26 payload=jobs_pb2.SendRawPushNotificationPayload( 

27 data=json.dumps( 

28 { 

29 "title": title[:500], 

30 "body": body[:2000], 

31 "icon": icon or urls.icon_url(), 

32 "url": url, 

33 } 

34 ).encode("utf8"), 

35 push_notification_subscription_id=push_notification_subscription_id, 

36 ttl=ttl, 

37 ), 

38 ) 

39 

40 

41def _push_to_user(user_id, **kwargs): 

42 """ 

43 Same as above but for a given user 

44 """ 

45 with session_scope() as session: 

46 sub_ids = ( 

47 session.execute( 

48 select(PushNotificationSubscription.id) 

49 .where(PushNotificationSubscription.user_id == user_id) 

50 .where(PushNotificationSubscription.disabled_at > func.now()) 

51 ) 

52 .scalars() 

53 .all() 

54 ) 

55 for sub_id in sub_ids: 

56 push_to_subscription(sub_id, **kwargs) 

57 

58 

59def push_to_user(user_id, **kwargs): 

60 """ 

61 This indirection is so that this can be easily mocked. Not sure how to do it better :( 

62 """ 

63 _push_to_user(user_id, **kwargs)