Coverage for apps/kwai-api/src/kwai_api/v1/pages/endpoints.py: 86%

1"""Module for defining the pages endpoint.""" 

2 

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

4from kwai_bc_identity.users.user import UserEntity 

5from kwai_bc_portal.applications.application_db_repository import ( 

6 ApplicationDbRepository, 

7) 

8from kwai_bc_portal.applications.application_repository import ( 

9 ApplicationNotFoundException, 

10) 

11from kwai_bc_portal.create_page import CreatePage, CreatePageCommand 

12from kwai_bc_portal.delete_page import DeletePage, DeletePageCommand 

13from kwai_bc_portal.get_page import GetPage, GetPageCommand 

14from kwai_bc_portal.get_pages import GetPages, GetPagesCommand 

15from kwai_bc_portal.pages.page_db_repository import PageDbRepository 

16from kwai_bc_portal.pages.page_repository import PageNotFoundException 

17from kwai_bc_portal.update_page import UpdatePage, UpdatePageCommand 

18from kwai_core.domain.use_case import TextCommand 

19from kwai_core.domain.value_objects.owner import Owner 

20from kwai_core.json_api import PaginationModel 

21from pydantic import BaseModel, Field 

22 

23from kwai_api.dependencies import create_database, get_current_user, get_optional_user 

24from kwai_api.v1.pages.presenters import JsonApiPagePresenter, JsonApiPagesPresenter 

25from kwai_api.v1.pages.schemas import PageDocument, PagesDocument 

26 

27 

28router = APIRouter() 

29 

30 

31class PageFilter(BaseModel): 

32 """Define the JSON:API filter for pages.""" 

33 

34 application: str | None = Field(Query(default=None, alias="filter[application]")) 

35 enabled: bool = Field(Query(default=True, alias="filter[enabled]")) 

36 

37 

38@router.get("/pages") 

39async def get_pages( 

40 pagination: PaginationModel = Depends(PaginationModel), 

41 page_filter: PageFilter = Depends(PageFilter), 

42 db=Depends(create_database), 

43 user: UserEntity | None = Depends(get_optional_user), 

44) -> PagesDocument: 

45 """Get pages.""" 

46 if user is None and not page_filter.enabled: 

47 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED) 

48 

49 command = GetPagesCommand( 

50 offset=pagination.offset or 0, 

51 limit=pagination.limit, 

52 application=page_filter.application, 

53 enabled=page_filter.enabled, 

54 ) 

55 presenter = JsonApiPagesPresenter(user) 

56 await GetPages(PageDbRepository(db), presenter).execute(command) 

57 return presenter.get_document() 

58 

59 

60@router.get("/pages/{id}") 

61async def get_page( 

62 id: int, 

63 db=Depends(create_database), 

64 user: UserEntity | None = Depends(get_optional_user), 

65) -> PageDocument: 

66 """Get page.""" 

67 command = GetPageCommand(id=id) 

68 presenter = JsonApiPagePresenter(user) 

69 await GetPage(PageDbRepository(db), presenter).execute(command) 

70 return presenter.get_document() 

71 

72 

73@router.post("/pages", status_code=status.HTTP_201_CREATED) 

74async def create_page( 

75 resource: PageDocument, 

76 db=Depends(create_database), 

77 user: UserEntity = Depends(get_current_user), 

78) -> PageDocument: 

79 """Create a page.""" 

80 if ( 

81 resource.data.relationships is None 

82 or resource.data.relationships.application is None 

83 or resource.data.relationships.application.data is None 

84 or resource.data.relationships.application.data.id is None 

85 ): 

86 raise HTTPException( 

87 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

88 detail="Missing application relationship", 

89 ) 

90 command = CreatePageCommand( 

91 enabled=resource.data.attributes.enabled, 

92 texts=[ 

93 TextCommand( 

94 locale=text.locale, 

95 format=text.format, 

96 title=text.title, 

97 summary=text.original_summary, 

98 content=text.original_content or "", 

99 ) 

100 for text in resource.data.attributes.texts 

101 ], 

102 application=int(resource.data.relationships.application.data.id), 

103 priority=resource.data.attributes.priority, 

104 remark=resource.data.attributes.remark, 

105 ) 

106 presenter = JsonApiPagePresenter(user) 

107 

108 try: 

109 await CreatePage( 

110 PageDbRepository(db), 

111 ApplicationDbRepository(db), 

112 Owner(id=user.id, uuid=user.uuid, name=user.name), 

113 presenter, 

114 ).execute(command) 

115 except ApplicationNotFoundException as ex: 

116 raise HTTPException( 

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

118 ) from ex 

119 

120 return presenter.get_document() 

121 

122 

123@router.patch("/pages/{id}") 

124async def update_page( 

125 id: int, 

126 resource: PageDocument, 

127 db=Depends(create_database), 

128 user: UserEntity = Depends(get_current_user), 

129) -> PageDocument: 

130 """Update a page.""" 

131 if ( 

132 resource.data.relationships is None 

133 or resource.data.relationships.application is None 

134 or resource.data.relationships.application.data is None 

135 or resource.data.relationships.application.data.id is None 

136 ): 

137 raise HTTPException( 

138 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

139 detail="Missing application relationship", 

140 ) 

141 

142 command = UpdatePageCommand( 

143 id=id, 

144 enabled=resource.data.attributes.enabled, 

145 texts=[ 

146 TextCommand( 

147 locale=text.locale, 

148 format=text.format, 

149 title=text.title, 

150 summary=text.original_summary, 

151 content=text.original_content or "", 

152 ) 

153 for text in resource.data.attributes.texts 

154 ], 

155 application=int(resource.data.relationships.application.data.id), 

156 priority=resource.data.attributes.priority, 

157 remark=resource.data.attributes.remark, 

158 ) 

159 presenter = JsonApiPagePresenter(user) 

160 try: 

161 await UpdatePage( 

162 PageDbRepository(db), 

163 ApplicationDbRepository(db), 

164 Owner(id=user.id, uuid=user.uuid, name=user.name), 

165 presenter, 

166 ).execute(command) 

167 except ApplicationNotFoundException as ex: 

168 raise HTTPException( 

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

170 ) from ex 

171 

172 return presenter.get_document() 

173 

174 

175@router.delete( 

176 "/pages/{id}", 

177 status_code=status.HTTP_200_OK, 

178 responses={status.HTTP_404_NOT_FOUND: {"description": "Page was not found."}}, 

179) 

180async def delete_news_item( 

181 id: int, 

182 db=Depends(create_database), 

183 user: UserEntity = Depends(get_current_user), 

184): 

185 """Delete a page.""" 

186 command = DeletePageCommand(id=id) 

187 

188 try: 

189 await DeletePage(PageDbRepository(db)).execute(command) 

190 except PageNotFoundException as ex: 

191 raise HTTPException( 

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

193 ) from ex