Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/invite_user.py: 100%

1"""Module for the invite user use case.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai_core.domain.exceptions import UnprocessableException 

6from kwai_core.domain.presenter import Presenter 

7from kwai_core.domain.value_objects.email_address import EmailAddress 

8from kwai_core.domain.value_objects.name import Name 

9from kwai_core.domain.value_objects.timestamp import Timestamp 

10from kwai_core.domain.value_objects.unique_id import UniqueId 

11from kwai_core.events.publisher import Publisher 

12 

13from kwai_bc_identity.user_invitations.user_invitation import UserInvitationEntity 

14from kwai_bc_identity.user_invitations.user_invitation_events import ( 

15 UserInvitationCreatedEvent, 

16) 

17from kwai_bc_identity.user_invitations.user_invitation_repository import ( 

18 UserInvitationRepository, 

19) 

20from kwai_bc_identity.users.user import UserEntity 

21from kwai_bc_identity.users.user_repository import ( 

22 UserNotFoundException, 

23 UserRepository, 

24) 

25 

26 

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

28class InviteUserCommand: 

29 """Input for the use case. 

30 

31 [InviteUser][kwai_bc_identity.invite_user.InviteUser] 

32 

33 Attributes: 

34 first_name: Firstname of the invited 

35 last_name: Lastname of the invited 

36 email: Email address of the invited 

37 expiration_in_days: After how many days will the invitation expire? 

38 remark: A remark about this invitation 

39 """ 

40 

41 first_name: str 

42 last_name: str 

43 email: str 

44 expiration_in_days: int = 7 

45 remark: str = "" 

46 

47 

48class InviteUser: 

49 """Invite user use case. 

50 

51 This use case will try to create an invitation for a user. 

52 """ 

53 

54 def __init__( 

55 self, 

56 user: UserEntity, 

57 user_repo: UserRepository, 

58 user_invitation_repo: UserInvitationRepository, 

59 presenter: Presenter[UserInvitationEntity], 

60 publisher: Publisher, 

61 ): 

62 """Initialize the use case. 

63 

64 Args: 

65 user: The inviter. 

66 user_repo: The repository to check if the email address is still free. 

67 user_invitation_repo: The repository for creating an invitation. 

68 presenter: A presenter for a user invitation. 

69 publisher: A publisher for publishing the user invitation created event. 

70 """ 

71 self._user = user 

72 self._user_repo = user_repo 

73 self._user_invitation_repo = user_invitation_repo 

74 self._presenter = presenter 

75 self._publisher = publisher 

76 

77 async def execute(self, command: InviteUserCommand) -> None: 

78 """Execute the use case. 

79 

80 Args: 

81 command: The input for this use case. 

82 

83 Raises: 

84 UnprocessableException: raised when the email address is already in use or 

85 when there are still pending invitations for the email address. 

86 """ 

87 email_address = EmailAddress(command.email) 

88 try: 

89 await self._user_repo.get_user_by_email(email_address) 

90 raise UnprocessableException(f"{command.email} is already used.") 

91 except UserNotFoundException: 

92 pass 

93 

94 query = ( 

95 self._user_invitation_repo.create_query() 

96 .filter_by_email(email_address) 

97 .filter_active() 

98 .filter_not_expired(Timestamp.create_now()) 

99 ) 

100 if await query.count() > 0: 

101 raise UnprocessableException( 

102 f"There are still pending invitations for {command.email}" 

103 ) 

104 

105 invitation = await self._user_invitation_repo.create( 

106 UserInvitationEntity( 

107 email=email_address, 

108 name=Name(first_name=command.first_name, last_name=command.last_name), 

109 uuid=UniqueId.generate(), 

110 expired_at=Timestamp.create_with_delta(days=command.expiration_in_days), 

111 user=self._user, 

112 ) 

113 ) 

114 

115 await self._publisher.publish( 

116 UserInvitationCreatedEvent(uuid=str(invitation.uuid)) 

117 ) 

118 

119 self._presenter.present(invitation)