Coverage for src/kwai/modules/identity/get_invitations.py: 100%

1"""Implement the use case: get user invitations.""" 

2 

3from dataclasses import dataclass 

4from typing import AsyncIterator 

5 

6from kwai.modules.identity.user_invitations.user_invitation import UserInvitationEntity 

7from kwai.modules.identity.user_invitations.user_invitation_repository import ( 

8 UserInvitationRepository, 

9) 

10 

11 

12@dataclass(kw_only=True, frozen=True, slots=True) 

13class GetInvitationsCommand: 

14 """Input for the use case. 

15 

16 [GetInvitations][kwai.modules.identity.get_invitations.GetInvitations] 

17 

18 Attributes: 

19 offset: Offset to use. Default is None. 

20 limit: The max. number of elements to return. Default is None, which means all. 

21 """ 

22 

23 offset: int | None = None 

24 limit: int | None = None 

25 active: bool = True 

26 

27 

28class GetInvitations: 

29 """Implementation of the use case. 

30 

31 Use this use case for getting user invitations. 

32 """ 

33 

34 def __init__(self, user_invitation_repo: UserInvitationRepository): 

35 """Initialize the use case. 

36 

37 Args: 

38 user_invitation_repo: A repository for getting the user invitations. 

39 """ 

40 self._user_invitation_repo = user_invitation_repo 

41 

42 async def execute( 

43 self, command: GetInvitationsCommand 

44 ) -> tuple[int, AsyncIterator[UserInvitationEntity]]: 

45 """Execute the use case. 

46 

47 Args: 

48 command: The input for this use case. 

49 

50 Returns: 

51 A tuple with the number of entities and an iterator for invitation entities. 

52 """ 

53 query = self._user_invitation_repo.create_query() 

54 if command.active: 

55 query = query.filter_active() 

56 

57 return ( 

58 await query.count(), 

59 self._user_invitation_repo.get_all( 

60 query=query, offset=command.offset, limit=command.limit 

61 ), 

62 )