Coverage for chalicelib/core/notifications.py: 56%

1import json 

2 

3from cachetools import TTLCache, cached 

4 

5from chalicelib.utils import pg_client, helper 

6from chalicelib.utils.TimeUTC import TimeUTC 

7 

8 

9def get_all(tenant_id, user_id): 

10 with pg_client.PostgresClient() as cur: 

11 cur.execute( 

12 cur.mogrify(""" \ 

13 SELECT notifications.*, 

14 user_viewed_notifications.notification_id NOTNULL AS viewed 

15 FROM public.notifications 

16 LEFT JOIN (SELECT notification_id 

17 FROM public.user_viewed_notifications 

18 WHERE user_viewed_notifications.user_id = %(user_id)s) AS user_viewed_notifications 

19 USING (notification_id) 

20 WHERE notifications.user_id IS NULL 

21 OR notifications.user_id = %(user_id)s 

22 ORDER BY created_at DESC LIMIT 100;""", 

23 {"user_id": user_id}) 

24 ) 

25 rows = helper.list_to_camel_case(cur.fetchall()) 

26 for r in rows: 26 ↛ 27line 26 didn't jump to line 27 because the loop on line 26 never started

27 r["createdAt"] = TimeUTC.datetime_to_timestamp(r["createdAt"]) 

28 return rows 

29 

30 

31cache = TTLCache(maxsize=1000, ttl=600) 

32 

33 

34@cached(cache) 

35def get_all_count(tenant_id, user_id): 

36 with pg_client.PostgresClient() as cur: 

37 cur.execute( 

38 cur.mogrify(""" \ 

39 SELECT COALESCE(COUNT(notifications.*), 0) AS count 

40 FROM public.notifications 

41 LEFT JOIN (SELECT notification_id 

42 FROM public.user_viewed_notifications 

43 WHERE user_viewed_notifications.user_id = %(user_id)s) AS user_viewed_notifications USING (notification_id) 

44 WHERE (notifications.user_id IS NULL 

45 OR notifications.user_id =%(user_id)s) 

46 AND user_viewed_notifications.notification_id IS NULL;""", 

47 {"user_id": user_id}) 

48 ) 

49 row = cur.fetchone() 

50 return row 

51 

52 

53def view_notification(user_id, notification_ids=[], tenant_id=None, startTimestamp=None, endTimestamp=None): 

54 if len(notification_ids) == 0 and endTimestamp is None: 

55 return False 

56 if startTimestamp is None: 

57 startTimestamp = 0 

58 notification_ids = [(user_id, id) for id in notification_ids] 

59 with pg_client.PostgresClient() as cur: 

60 if len(notification_ids) > 0: 

61 cur.executemany( 

62 "INSERT INTO public.user_viewed_notifications(user_id, notification_id) VALUES (%s,%s) ON CONFLICT DO NOTHING;", 

63 notification_ids) 

64 else: 

65 query = """INSERT INTO public.user_viewed_notifications(user_id, notification_id) 

66 SELECT %(user_id)s AS user_id, notification_id 

67 FROM public.notifications 

68 WHERE (user_id IS NULL OR user_id = %(user_id)s) 

69 AND EXTRACT(EPOCH FROM created_at) * 1000 >= (%(startTimestamp)s) 

70 AND EXTRACT(EPOCH FROM created_at) * 1000 <= (%(endTimestamp)s + 1000) ON CONFLICT DO NOTHING;""" 

71 params = {"user_id": user_id, "startTimestamp": startTimestamp, 

72 "endTimestamp": endTimestamp} 

73 # print('-------------------') 

74 # print(cur.mogrify(query, params)) 

75 cur.execute(cur.mogrify(query, params)) 

76 return True 

77 

78 

79def create(notifications): 

80 if len(notifications) == 0: 

81 return [] 

82 with pg_client.PostgresClient() as cur: 

83 values = [] 

84 for n in notifications: 

85 clone = dict(n) 

86 if "userId" not in clone: 

87 clone["userId"] = None 

88 if "options" not in clone: 

89 clone["options"] = '{}' 

90 else: 

91 clone["options"] = json.dumps(clone["options"]) 

92 values.append( 

93 cur.mogrify( 

94 "(%(userId)s, %(title)s, %(description)s, %(buttonText)s, %(buttonUrl)s, %(imageUrl)s,%(options)s)", 

95 clone).decode('UTF-8') 

96 ) 

97 cur.execute( 

98 f"""INSERT INTO public.notifications(user_id, title, description, button_text, button_url, image_url, options)  

99 VALUES {",".join(values)} RETURNING *;""") 

100 rows = helper.list_to_camel_case(cur.fetchall()) 

101 for r in rows: 

102 r["createdAt"] = TimeUTC.datetime_to_timestamp(r["createdAt"]) 

103 r["viewed"] = False 

104 return rows