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

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

2from dataclasses import dataclass 

3from typing import AsyncIterator 

4 

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

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

7 UserInvitationRepository, 

8) 

9 

10 

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

12class GetInvitationsCommand: 

13 """Input for the use case. 

14 

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

16 

17 Attributes: 

18 offset: Offset to use. Default is None. 

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

20 """ 

21 

22 offset: int | None = None 

23 limit: int | None = None 

24 

25 

26class GetInvitations: 

27 """Implementation of the use case. 

28 

29 Use this use case for getting user invitations. 

30 """ 

31 

32 def __init__(self, user_invitation_repo: UserInvitationRepository): 

33 """Initialize the use case. 

34 

35 Args: 

36 user_invitation_repo: A repository for getting the user invitations. 

37 """ 

38 self._user_invitation_repo = user_invitation_repo 

39 

40 async def execute( 

41 self, command: GetInvitationsCommand 

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

43 """Execute the use case. 

44 

45 Args: 

46 command: The input for this use case. 

47 

48 Returns: 

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

50 """ 

51 query = self._user_invitation_repo.create_query() 

52 return ( 

53 await query.count(), 

54 self._user_invitation_repo.get_all( 

55 query=query, offset=command.offset, limit=command.limit 

56 ), 

57 )