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 

26 

27class GetInvitations: 

28 """Implementation of the use case. 

29 

30 Use this use case for getting user invitations. 

31 """ 

32 

33 def __init__(self, user_invitation_repo: UserInvitationRepository): 

34 """Initialize the use case. 

35 

36 Args: 

37 user_invitation_repo: A repository for getting the user invitations. 

38 """ 

39 self._user_invitation_repo = user_invitation_repo 

40 

41 async def execute( 

42 self, command: GetInvitationsCommand 

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

44 """Execute the use case. 

45 

46 Args: 

47 command: The input for this use case. 

48 

49 Returns: 

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

51 """ 

52 query = self._user_invitation_repo.create_query() 

53 return ( 

54 await query.count(), 

55 self._user_invitation_repo.get_all( 

56 query=query, offset=command.offset, limit=command.limit 

57 ), 

58 )