Coverage for app / backend / src / couchers / helpers / badges.py: 100%

1from sqlalchemy import exists, select 

2from sqlalchemy.orm import Session 

3from sqlalchemy.sql import delete 

4 

5from couchers.context import make_background_user_context 

6from couchers.models import UserBadge 

7from couchers.models.notifications import NotificationTopicAction 

8from couchers.notifications.notify import notify 

9from couchers.proto import notification_data_pb2 

10from couchers.resources import get_badge_dict 

11 

12 

13def user_add_badge(session: Session, user_id: int, badge_id: str, do_notify: bool = True) -> None: 

14 badge = get_badge_dict()[badge_id] 

15 already_has_badge = session.execute( 

16 select(exists().where(UserBadge.user_id == user_id, UserBadge.badge_id == badge_id)) 

17 ).scalar() 

18 if already_has_badge: 

19 return 

20 session.add(UserBadge(user_id=user_id, badge_id=badge_id)) 

21 session.flush() 

22 if do_notify: 

23 context = make_background_user_context(user_id=user_id) 

24 notify( 

25 session, 

26 user_id=user_id, 

27 topic_action=NotificationTopicAction.badge__add, 

28 key=badge.id, 

29 data=notification_data_pb2.BadgeAdd( 

30 badge_id=badge.id, 

31 badge_name=context.localization.localize_string(f"badges.{badge.id}_name"), 

32 badge_description=context.localization.localize_string(f"badges.{badge.id}_description"), 

33 ), 

34 ) 

35 session.commit() 

36 

37 

38def user_remove_badge(session: Session, user_id: int, badge_id: str) -> None: 

39 badge = get_badge_dict()[badge_id] 

40 session.execute(delete(UserBadge).where(UserBadge.user_id == user_id, UserBadge.badge_id == badge.id)) 

41 session.flush() 

42 context = make_background_user_context(user_id=user_id) 

43 notify( 

44 session, 

45 user_id=user_id, 

46 topic_action=NotificationTopicAction.badge__remove, 

47 key=badge.id, 

48 data=notification_data_pb2.BadgeRemove( 

49 badge_id=badge.id, 

50 badge_name=context.localization.localize_string(f"badges.{badge.id}_name"), 

51 badge_description=context.localization.localize_string(f"badges.{badge.id}_description"), 

52 ), 

53 ) 

54 session.commit()