Coverage for apps/kwai-api/src/kwai_api/v1/pages/presenters.py: 92%

1"""Module that defines presenters for the portal/pages endpoint.""" 

2 

3from typing import Self 

4 

5from kwai_bc_identity.users.user import UserEntity 

6from kwai_bc_portal.domain.application import ApplicationEntity 

7from kwai_bc_portal.domain.page import PageEntity 

8from kwai_core.domain.exceptions import NotAllowedException 

9from kwai_core.domain.presenter import AsyncPresenter, IterableResult, Presenter 

10from kwai_core.json_api import JsonApiPresenter, Meta, Relationship, ResourceMeta 

11 

12from kwai_api.converter import MarkdownConverter 

13from kwai_api.schemas.resources import ApplicationResourceIdentifier 

14from kwai_api.v1.pages.schemas import ( 

15 PageApplicationAttributes, 

16 PageApplicationDocument, 

17 PageApplicationResource, 

18 PageAttributes, 

19 PageDocument, 

20 PageRelationships, 

21 PageResource, 

22 PagesDocument, 

23 PageText, 

24) 

25 

26 

27class JsonApiPageApplicationPresenter( 

28 JsonApiPresenter[PageApplicationDocument], Presenter[ApplicationEntity] 

29): 

30 """A presenter that transforms a news application into a JSON:API document.""" 

31 

32 def present(self, news_application: ApplicationEntity) -> Self: 

33 self._document = PageApplicationDocument( 

34 data=PageApplicationResource( 

35 id=str(news_application.id), 

36 attributes=PageApplicationAttributes( 

37 name=news_application.name, title=news_application.title 

38 ), 

39 ) 

40 ) 

41 return self 

42 

43 

44class JsonApiPagePresenter(JsonApiPresenter[PageDocument], Presenter[PageEntity]): 

45 """A presenter that transform an entity into a JSON:API document.""" 

46 

47 def __init__(self, user: UserEntity | None): 

48 super().__init__() 

49 self._user = user 

50 

51 def present(self, page: PageEntity) -> Self: 

52 # Only a know user is allowed to see a disabled page. 

53 if not self._user and not page.enabled: 

54 raise NotAllowedException(f"Not allowed to view the page {page.id}") 

55 

56 page_application_document = ( 

57 JsonApiPageApplicationPresenter().present(page.application).get_document() 

58 ) 

59 

60 self._document = PageDocument( 

61 data=PageResource( 

62 id=str(page.id), 

63 meta=ResourceMeta( 

64 created_at=str(page.traceable_time.created_at), 

65 updated_at=str(page.traceable_time.updated_at), 

66 ), 

67 attributes=PageAttributes( 

68 enabled=page.enabled, 

69 priority=page.priority, 

70 remark=page.remark or "", 

71 texts=[ 

72 PageText( 

73 locale=text.locale.value, 

74 format=text.format.value, 

75 title=text.title, 

76 summary=MarkdownConverter().convert(text.summary), 

77 content=MarkdownConverter().convert(text.content) 

78 if text.content 

79 else None, 

80 original_summary=text.summary, 

81 original_content=text.content, 

82 ) 

83 for text in page.texts 

84 ], 

85 ), 

86 relationships=PageRelationships( 

87 application=Relationship[ApplicationResourceIdentifier]( 

88 data=ApplicationResourceIdentifier(id=str(page.application.id)) 

89 ) 

90 ), 

91 ), 

92 included={page_application_document.data}, 

93 ) 

94 return self 

95 

96 

97class JsonApiPagesPresenter( 

98 JsonApiPresenter[PagesDocument], AsyncPresenter[IterableResult[PageEntity]] 

99): 

100 """A presenter that transfer an iterable list of news items into a JSON:API document.""" 

101 

102 def __init__(self, user: UserEntity | None): 

103 super().__init__() 

104 self._user = user 

105 

106 async def present(self, result: IterableResult[PageEntity]) -> Self: 

107 self._document = PagesDocument( 

108 meta=Meta(count=result.count, offset=result.offset, limit=result.limit), 

109 data=[], 

110 ) 

111 async for page in result.iterator: 

112 try: 

113 page_document = ( 

114 JsonApiPagePresenter(self._user).present(page).get_document() 

115 ) 

116 self._document.data.append(page_document.data) 

117 self._document.included |= page_document.included 

118 except NotAllowedException: 

119 self._document.meta.count -= 1 

120 

121 return self