Coverage for src/couchers/notifications/notify.py: 100%

17 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 16:44 +0000

1import logging 

2 

3from google.protobuf import empty_pb2 

4 

5from couchers.db import session_scope 

6from couchers.jobs.enqueue import queue_job 

7from couchers.models import Notification 

8from couchers.notifications.utils import enum_from_topic_action 

9from proto.internal import jobs_pb2 

10 

11logger = logging.getLogger(__name__) 

12 

13 

14def notify( 

15 *, 

16 user_id, 

17 topic_action, 

18 key="", 

19 data=None, 

20): 

21 """ 

22 Queues a notification given the notification and a target, i.e. a tuple (user_id, topic, key), and an action. 

23 

24 Notifications are sent to user identified by user_id, and are collapsed/grouped based on the combination of 

25 (topic, key). 

26 

27 For example, topic may be "chat" for a group chat/direct message, and the key might be the chat id; so that messages 

28 in the same group chat show up in one group. 

29 

30 The action is a simple identifier describing the action that caused the notification. For the above example, the 

31 action might be "add_admin" if the notification was caused by another user adding an admin into the gorup chat. 

32 

33 Each different notification type should have its own action. 

34 """ 

35 logger.info(f"Generating notification of type {topic_action} for user {user_id}") 

36 topic, action = topic_action.split(":") 

37 with session_scope() as session: 

38 notification = Notification( 

39 user_id=user_id, 

40 topic_action=enum_from_topic_action[topic, action], 

41 key=key, 

42 data=(data or empty_pb2.Empty()).SerializeToString(), 

43 ) 

44 session.add(notification) 

45 session.flush() 

46 notification_id = notification.id 

47 

48 queue_job( 

49 job_type="handle_notification", 

50 payload=jobs_pb2.HandleNotificationPayload( 

51 notification_id=notification_id, 

52 ), 

53 )