Coverage for kwai/api/v1/auth/endpoints/user_invitations.py: 79%

1"""Module that implements invitations endpoints.""" 

2 

3from fastapi import APIRouter, Depends, HTTPException, Response, status 

4from loguru import logger 

5 

6from kwai.api.dependencies import deps, get_current_user 

7from kwai.api.schemas.user_invitation import UserInvitationResource 

8from kwai.core.db.database import Database 

9from kwai.core.domain.exceptions import UnprocessableException 

10from kwai.core.domain.value_objects.email_address import InvalidEmailException 

11from kwai.core.events.bus import Bus 

12from kwai.core.json_api import Meta, PaginationModel 

13from kwai.modules.identity.delete_user_invitation import ( 

14 DeleteUserInvitation, 

15 DeleteUserInvitationCommand, 

16) 

17from kwai.modules.identity.get_invitations import GetInvitations, GetInvitationsCommand 

18from kwai.modules.identity.get_user_invitation import ( 

19 GetUserInvitation, 

20 GetUserInvitationCommand, 

21) 

22from kwai.modules.identity.invite_user import InviteUser, InviteUserCommand 

23from kwai.modules.identity.user_invitations.user_invitation_db_repository import ( 

24 UserInvitationDbRepository, 

25) 

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

27 UserInvitationNotFoundException, 

28) 

29from kwai.modules.identity.users.user import UserEntity 

30from kwai.modules.identity.users.user_db_repository import UserDbRepository 

31 

32router = APIRouter() 

33 

34 

35@router.post("/invitations") 

36async def create_user_invitation( 

37 resource: UserInvitationResource.get_document_model(), 

38 db=deps.depends(Database), 

39 user: UserEntity = Depends(get_current_user), 

40 bus=deps.depends(Bus), 

41) -> UserInvitationResource.get_document_model(): 

42 """Create a user invitation.""" 

43 command = InviteUserCommand( 

44 first_name=resource.data.attributes.first_name, 

45 last_name=resource.data.attributes.last_name, 

46 email=resource.data.attributes.email, 

47 remark=resource.data.attributes.remark, 

48 ) 

49 

50 try: 

51 invitation = await InviteUser( 

52 user, UserDbRepository(db), UserInvitationDbRepository(db), bus 

53 ).execute(command) 

54 except InvalidEmailException as exc: 

55 raise HTTPException( 

56 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

57 detail="Invalid email address", 

58 ) from exc 

59 except UnprocessableException as ex: 

60 logger.warning(f"User invitation could not be processed: {ex}") 

61 raise HTTPException( 

62 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ex) 

63 ) from ex 

64 

65 return UserInvitationResource(invitation).serialize() 

66 

67 

68@router.delete( 

69 "/invitations/{uuid}", 

70 summary="Delete a user invitation", 

71 status_code=status.HTTP_200_OK, 

72 response_class=Response, 

73) 

74async def delete_user_invitation( 

75 uuid: str, 

76 db=deps.depends(Database), 

77 user: UserEntity = Depends(get_current_user), 

78): 

79 """Delete the user invitation with the given unique id.""" 

80 command = DeleteUserInvitationCommand(uuid=uuid) 

81 try: 

82 await DeleteUserInvitation(UserInvitationDbRepository(db)).execute(command) 

83 except UserInvitationNotFoundException as ex: 

84 raise HTTPException( 

85 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

86 ) from ex 

87 except ValueError as ex: 

88 raise HTTPException( 

89 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ex) 

90 ) from ex 

91 

92 

93@router.get("/invitations") 

94async def get_user_invitations( 

95 pagination: PaginationModel = Depends(PaginationModel), 

96 db=deps.depends(Database), 

97 user: UserEntity = Depends(get_current_user), 

98) -> UserInvitationResource.get_document_model(): 

99 """Get all user invitations.""" 

100 command = GetInvitationsCommand(offset=pagination.offset, limit=pagination.limit) 

101 count, invitation_iterator = await GetInvitations( 

102 UserInvitationDbRepository(db) 

103 ).execute(command) 

104 

105 document = UserInvitationResource.serialize_list( 

106 [UserInvitationResource(invitation) async for invitation in invitation_iterator] 

107 ) 

108 document.meta = Meta(count=count) 

109 return document 

110 

111 

112@router.get("/invitations/{uuid}") 

113async def get_user_invitation( 

114 uuid: str, 

115 db=deps.depends(Database), 

116 user: UserEntity = Depends(get_current_user), 

117) -> UserInvitationResource.get_document_model(): 

118 """Get the user invitation with the given unique id.""" 

119 command = GetUserInvitationCommand(uuid=uuid) 

120 try: 

121 invitation = await GetUserInvitation(UserInvitationDbRepository(db)).execute( 

122 command 

123 ) 

124 except UserInvitationNotFoundException as ex: 

125 raise HTTPException( 

126 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

127 ) from ex 

128 

129 return UserInvitationResource(invitation).serialize()