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

1from sqlalchemy.orm import Session 

2from sqlalchemy.sql import delete 

3 

4from couchers.context import make_background_user_context 

5from couchers.models import UserBadge 

6from couchers.models.notifications import NotificationTopicAction 

7from couchers.notifications.notify import notify 

8from couchers.proto import notification_data_pb2 

9from couchers.resources import get_badge_dict 

10 

11 

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

13 badge = get_badge_dict()[badge_id] 

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

15 session.flush() 

16 if do_notify: 

17 context = make_background_user_context(user_id=user_id) 

18 notify( 

19 session, 

20 user_id=user_id, 

21 topic_action=NotificationTopicAction.badge__add, 

22 key=badge.id, 

23 data=notification_data_pb2.BadgeAdd( 

24 badge_id=badge.id, 

25 badge_name=context.get_localized_string(f"badges.{badge.id}_name"), 

26 badge_description=context.get_localized_string(f"badges.{badge.id}_description"), 

27 ), 

28 ) 

29 session.commit() 

30 

31 

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

33 badge = get_badge_dict()[badge_id] 

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

35 session.flush() 

36 context = make_background_user_context(user_id=user_id) 

37 notify( 

38 session, 

39 user_id=user_id, 

40 topic_action=NotificationTopicAction.badge__remove, 

41 key=badge.id, 

42 data=notification_data_pb2.BadgeRemove( 

43 badge_id=badge.id, 

44 badge_name=context.get_localized_string(f"badges.{badge.id}_name"), 

45 badge_description=context.get_localized_string(f"badges.{badge.id}_description"), 

46 ), 

47 ) 

48 session.commit()