Coverage for polar/notification_recipient/repository.py: 37%

1from collections.abc import Sequence 1 ctx1a

2from uuid import UUID 1 ctx1a

3 

4from polar.kit.repository import ( 1 ctx1a

5 RepositoryBase, 

6 RepositorySoftDeletionIDMixin, 

7 RepositorySoftDeletionMixin, 

8) 

9from polar.kit.repository.base import Options 1 ctx1a

10from polar.models.notification_recipient import NotificationRecipient 1 ctx1a

11 

12from .schemas import NotificationRecipientPlatform 1 ctx1a

13 

14 

15class NotificationRecipientRepository( 1 ctx1a

16 RepositorySoftDeletionIDMixin[NotificationRecipient, UUID], 

17 RepositorySoftDeletionMixin[NotificationRecipient], 

18 RepositoryBase[NotificationRecipient], 

19): 

20 model = NotificationRecipient 1 ctx1a

21 

22 async def delete( 1 ctx1a

23 self, notification_recipient_id: UUID, user_id: UUID, *, flush: bool = False 

24 ) -> None: 

25 # First get the notification recipient 

26 statement = ( 

27 self.get_base_statement() 

28 .where(NotificationRecipient.id == notification_recipient_id) 

29 .where(NotificationRecipient.user_id == user_id) 

30 ) 

31 notification_recipient = await self.get_one_or_none(statement) 

32 

33 # If notification recipient exists, soft delete it 

34 if notification_recipient: 

35 await self.soft_delete(notification_recipient, flush=flush) 

36 

37 return None 

38 

39 async def list_by_user( 1 ctx1a

40 self, 

41 user_id: UUID, 

42 platform: NotificationRecipientPlatform | None, 

43 expo_push_token: str | None, 

44 *, 

45 options: Options = (), 

46 ) -> Sequence[NotificationRecipient]: 

47 statement = self.get_base_statement().where( 

48 NotificationRecipient.user_id == user_id 

49 ) 

50 

51 if expo_push_token: 

52 statement = statement.where( 

53 NotificationRecipient.expo_push_token == expo_push_token 

54 ) 

55 

56 if platform: 

57 statement = statement.where(NotificationRecipient.platform == platform) 

58 

59 return await self.get_all(statement.options(*options)) 

60 

61 async def get_by_expo_token( 1 ctx1a

62 self, expo_push_token: str 

63 ) -> NotificationRecipient | None: 

64 statement = self.get_base_statement().where( 

65 NotificationRecipient.expo_push_token == expo_push_token 

66 ) 

67 return await self.get_one_or_none(statement)