Coverage for kwai/modules/identity/user_invitations/user_invitation_repository.py: 100%

1"""Module that defines an interface for an invitation repository.""" 

2from abc import ABC, abstractmethod 

3from typing import AsyncIterator 

4 

5from kwai.core.domain.value_objects.unique_id import UniqueId 

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

7 UserInvitationEntity, 

8 UserInvitationIdentifier, 

9) 

10from kwai.modules.identity.user_invitations.user_invitation_query import ( 

11 UserInvitationQuery, 

12) 

13 

14 

15class UserInvitationNotFoundException(Exception): 

16 """Raised when the invitation could not be found.""" 

17 

18 

19class UserInvitationRepository(ABC): 

20 """An invitation repository interface.""" 

21 

22 @abstractmethod 

23 def create_query(self) -> UserInvitationQuery: 

24 """Create a UserInvitationQuery. 

25 

26 Returns: 

27 A query for user invitations. 

28 """ 

29 raise NotImplementedError 

30 

31 @abstractmethod 

32 async def get_all( 

33 self, 

34 query: UserInvitationQuery, 

35 limit: int | None = None, 

36 offset: int | None = None, 

37 ) -> AsyncIterator[UserInvitationEntity]: 

38 """Return all user invitations from the query. 

39 

40 Args: 

41 query: The prepared query. 

42 limit: The maximum number of entities to return. 

43 offset: Skip the offset rows before beginning to return entities. 

44 

45 Yields: 

46 A list of user invitation entities. 

47 """ 

48 raise NotImplementedError 

49 

50 @abstractmethod 

51 async def get_invitation_by_id( 

52 self, id_: UserInvitationIdentifier 

53 ) -> UserInvitationEntity: 

54 """Get an invitation using the id. 

55 

56 Args: 

57 id_(UserInvitationIdentifier): The id of the invitation to search for. 

58 """ 

59 raise NotImplementedError 

60 

61 @abstractmethod 

62 async def get_invitation_by_uuid(self, uuid: UniqueId) -> UserInvitationEntity: 

63 """Get an invitation using the unique id. 

64 

65 Args: 

66 uuid(UniqueId): The unique id to use for searching the invitation. 

67 """ 

68 raise NotImplementedError 

69 

70 @abstractmethod 

71 async def create(self, invitation: UserInvitationEntity) -> UserInvitationEntity: 

72 """Create a new invitation. 

73 

74 Args: 

75 invitation(UserInvitationEntity): The invitation to create. 

76 """ 

77 raise NotImplementedError 

78 

79 @abstractmethod 

80 async def update(self, invitation: UserInvitationEntity) -> None: 

81 """Update an existing invitation. 

82 

83 Args: 

84 invitation(UserInvitationEntity): The invitation to update. 

85 """ 

86 raise NotImplementedError 

87 

88 @abstractmethod 

89 async def delete(self, invitation: UserInvitationEntity) -> None: 

90 """Delete the invitation. 

91 

92 Args: 

93 invitation(UserInvitationEntity): The invitation to delete. 

94 """ 

95 raise NotImplementedError