Coverage for apps/kwai-api/src/kwai_api/v1/auth/endpoints/user_invitations.py: 73%
70 statements
« prev ^ index » next coverage.py v7.11.0, created at 2024-01-01 00:00 +0000
1"""Module that implements invitations endpoints."""
3from typing import Annotated
5from fastapi import APIRouter, Depends, HTTPException, status
6from kwai_bc_identity.delete_user_invitation import (
7 DeleteUserInvitation,
8 DeleteUserInvitationCommand,
9)
10from kwai_bc_identity.get_invitations import GetInvitations, GetInvitationsCommand
11from kwai_bc_identity.get_user_invitation import (
12 GetUserInvitation,
13 GetUserInvitationCommand,
14)
15from kwai_bc_identity.invite_user import InviteUser, InviteUserCommand
16from kwai_bc_identity.recreate_user_invitation import (
17 RecreateUserInvitation,
18 RecreateUserInvitationCommand,
19)
20from kwai_bc_identity.user_invitations.user_invitation_db_repository import (
21 UserInvitationDbRepository,
22)
23from kwai_bc_identity.user_invitations.user_invitation_repository import (
24 UserInvitationNotFoundException,
25)
26from kwai_bc_identity.users.user import UserEntity
27from kwai_bc_identity.users.user_db_repository import UserDbRepository
28from kwai_core.db.database import Database
29from kwai_core.db.uow import UnitOfWork
30from kwai_core.domain.exceptions import UnprocessableException
31from kwai_core.domain.value_objects.email_address import InvalidEmailException
32from kwai_core.events.publisher import Publisher
33from kwai_core.json_api import PaginationModel
34from loguru import logger
36from kwai_api.dependencies import create_database, get_current_user, get_publisher
37from kwai_api.v1.auth.presenters import (
38 JsonApiUserInvitationPresenter,
39 JsonApiUserInvitationsPresenter,
40)
41from kwai_api.v1.auth.schemas.user_invitation import (
42 UserInvitationDocument,
43 UserInvitationsDocument,
44)
47router = APIRouter()
50@router.post(
51 "/invitations/{uuid}",
52 summary="Recreate a user invitation",
53 status_code=status.HTTP_201_CREATED,
54 responses={
55 201: {"description": "User invitation is created"},
56 401: {"description": "Not authorized."},
57 422: {"description": "User invitation could not be created"},
58 },
59)
60async def recreate_user_invitation(
61 uuid: str,
62 db: Annotated[Database, Depends(create_database)],
63 user: Annotated[UserEntity, Depends(get_current_user)],
64 publisher: Annotated[Publisher, Depends(get_publisher)],
65) -> UserInvitationDocument:
66 """Recreate a user invitation.
68 Use this API for resending a user invitation. The existing invitation
69 will expire.
70 """
71 command = RecreateUserInvitationCommand(uuid=uuid)
72 try:
73 async with UnitOfWork(db):
74 invitation = await RecreateUserInvitation(
75 user, UserDbRepository(db), UserInvitationDbRepository(db), publisher
76 ).execute(command)
77 except UnprocessableException as ex:
78 logger.warning(f"User invitation could not be processed: {ex}")
79 raise HTTPException(
80 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ex)
81 ) from ex
83 return UserInvitationDocument.create(invitation)
86@router.post(
87 "/invitations",
88 summary="Create a user invitation",
89 status_code=status.HTTP_201_CREATED,
90 responses={
91 201: {"description": "User invitation is created"},
92 401: {"description": "Not authorized."},
93 422: {"description": "User invitation could not be created"},
94 },
95)
96async def create_user_invitation(
97 user_invitation_document: UserInvitationDocument,
98 db=Depends(create_database),
99 user: UserEntity = Depends(get_current_user),
100 publisher=Depends(get_publisher),
101) -> UserInvitationDocument:
102 """Create a user invitation.
104 A wrong email address or a still pending user invitation will result in a 422
105 status code.
106 """
107 command = InviteUserCommand(
108 first_name=user_invitation_document.data.attributes.first_name,
109 last_name=user_invitation_document.data.attributes.last_name,
110 email=user_invitation_document.data.attributes.email,
111 remark=user_invitation_document.data.attributes.remark,
112 )
113 presenter = JsonApiUserInvitationPresenter()
114 try:
115 async with UnitOfWork(db):
116 await InviteUser(
117 user,
118 UserDbRepository(db),
119 UserInvitationDbRepository(db),
120 presenter,
121 publisher,
122 ).execute(command)
123 except InvalidEmailException as exc:
124 raise HTTPException(
125 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
126 detail="Invalid email address",
127 ) from exc
128 except UnprocessableException as ex:
129 logger.warning(f"User invitation could not be processed: {ex}")
130 raise HTTPException(
131 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ex)
132 ) from ex
134 return presenter.get_document()
137@router.delete(
138 "/invitations/{uuid}",
139 summary="Delete a user invitation",
140 status_code=status.HTTP_200_OK,
141 responses={
142 200: {"description": "User invitation is deleted."},
143 401: {"description": "Not authorized."},
144 404: {"description": "User invitation does not exist."},
145 422: {"description": "Invalid unique id passed for the user invitation."},
146 },
147)
148async def delete_user_invitation(
149 uuid: str,
150 db=Depends(create_database),
151 user: UserEntity = Depends(get_current_user),
152) -> None:
153 """Delete the user invitation with the given unique id."""
154 command = DeleteUserInvitationCommand(uuid=uuid)
155 try:
156 async with UnitOfWork(db):
157 await DeleteUserInvitation(UserInvitationDbRepository(db)).execute(command)
158 except UserInvitationNotFoundException as ex:
159 raise HTTPException(
160 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
161 ) from ex
162 except ValueError as ex:
163 raise HTTPException(
164 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ex)
165 ) from ex
168@router.get(
169 "/invitations",
170 summary="Get a list of user invitations",
171 responses={
172 200: {"description": "Ok."},
173 401: {"description": "Not authorized."},
174 },
175)
176async def get_user_invitations(
177 pagination: PaginationModel = Depends(PaginationModel),
178 db=Depends(create_database),
179 user: UserEntity = Depends(get_current_user),
180) -> UserInvitationsDocument:
181 """Get all user invitations.
183 Use the page[offset] and page[limit] query parameters to get a paginated result.
184 """
185 command = GetInvitationsCommand(offset=pagination.offset, limit=pagination.limit)
186 presenter = JsonApiUserInvitationsPresenter()
187 await GetInvitations(UserInvitationDbRepository(db), presenter).execute(command)
189 return presenter.get_document()
192@router.get(
193 "/invitations/{uuid}",
194 summary="Get a user invitation",
195 responses={200: {"description": "Ok."}, 401: {"description": "Not authorized."}},
196)
197async def get_user_invitation(
198 uuid: str,
199 db=Depends(create_database),
200 user: UserEntity = Depends(get_current_user),
201) -> UserInvitationDocument:
202 """Get the user invitation with the given unique id."""
203 command = GetUserInvitationCommand(uuid=uuid)
204 presenter = JsonApiUserInvitationPresenter()
205 try:
206 await GetUserInvitation(UserInvitationDbRepository(db), presenter).execute(
207 command
208 )
209 except UserInvitationNotFoundException as ex:
210 raise HTTPException(
211 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
212 ) from ex
214 return presenter.get_document()