Coverage for app/services/notifications.py: 86%

1import sqlalchemy 

2from app.core.exceptions import DatabaseQueryError 

3from app.db.models import Notification, UserCampaignRole 

4from fastapi import HTTPException 

5from sqlalchemy.exc import SQLAlchemyError 

6 

7 

8def get_user_notifications(limit, offset, user_participations, db): 

9 try: 

10 # make set of all world and campaign id's 

11 user_campaigns = {item.id for item in user_participations["campaign_info"]} 

12 user_worlds = {item.id for item in user_participations["world_info"]} 

13 

14 # get all notifications for worlds and campaigns the user is a part of 

15 world_notifications = db.query(Notification).filter(Notification.world_id.in_(user_worlds)).limit(limit).offset(offset) 

16 campaign_notifications = db.query(Notification).filter(Notification.campaign_id.in_(user_campaigns)).limit(limit).offset(offset) 

17 

18 return { 

19 "user_world_notifications": world_notifications, 

20 "user_campaign_notifications": campaign_notifications, 

21 } 

22 

23 except SQLAlchemyError as e: 

24 raise DatabaseQueryError()