Coverage for chalicelib/core/announcements.py: 74%

1from chalicelib.utils import pg_client 

2from chalicelib.utils import helper 

3from decouple import config 

4from chalicelib.utils.TimeUTC import TimeUTC 

5 

6 

7def get_all(user_id): 

8 with pg_client.PostgresClient() as cur: 

9 query = cur.mogrify(""" 

10 SELECT a.*, u.last >= (EXTRACT(EPOCH FROM a.created_at)*1000) AS viewed 

11 FROM public.announcements AS a, 

12 (SELECT COALESCE(CAST(data ->> 'lastAnnouncementView' AS bigint), 0) 

13 FROM public.users 

14 WHERE user_id = %(userId)s 

15 LIMIT 1) AS u(last) 

16 ORDER BY a.created_at DESC;""", 

17 {"userId": user_id}) 

18 cur.execute( 

19 query 

20 ) 

21 announcements = helper.list_to_camel_case(cur.fetchall()) 

22 for a in announcements: 22 ↛ 23line 22 didn't jump to line 23 because the loop on line 22 never started

23 a["createdAt"] = TimeUTC.datetime_to_timestamp(a["createdAt"]) 

24 if a["imageUrl"] is not None and len(a["imageUrl"]) > 0: 

25 a["imageUrl"] = config("announcement_url") + a["imageUrl"] 

26 return announcements 

27 

28 

29def view(user_id): 

30 with pg_client.PostgresClient() as cur: 

31 query = cur.mogrify(""" 

32 UPDATE public.users 

33 SET data=data || 

34 ('{"lastAnnouncementView":' || 

35 (EXTRACT(EPOCH FROM timezone('utc'::text, now())) * 1000)::bigint - 20 * 000 || 

36 '}')::jsonb 

37 WHERE user_id = %(userId)s;""", 

38 {"userId": user_id}) 

39 cur.execute( 

40 query 

41 ) 

42 return True