Coverage for bc/kwai-bc-portal/src/kwai_bc_portal/update_news_item.py: 90%

1"""Module for implementing the use case "Update News Item".""" 

2 

3from dataclasses import dataclass, replace 

4 

5from kwai_core.domain.presenter import Presenter 

6from kwai_core.domain.value_objects.owner import Owner 

7from kwai_core.domain.value_objects.period import Period 

8from kwai_core.domain.value_objects.text import DocumentFormat, Locale, LocaleText 

9from kwai_core.domain.value_objects.timestamp import Timestamp 

10 

11from kwai_bc_portal.applications.application_repository import ( 

12 ApplicationRepository, 

13) 

14from kwai_bc_portal.domain.application import ApplicationIdentifier 

15from kwai_bc_portal.domain.news_item import ( 

16 NewsItemEntity, 

17 NewsItemIdentifier, 

18 Promotion, 

19) 

20from kwai_bc_portal.news.news_item_repository import NewsItemRepository 

21from kwai_bc_portal.news_item_command import NewsItemCommand 

22 

23 

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

25class UpdateNewsItemCommand(NewsItemCommand): 

26 """Input for the use case "Update News Item".""" 

27 

28 id: int 

29 

30 

31class UpdateNewsItem: 

32 """Use case for updating a news item.""" 

33 

34 def __init__( 

35 self, 

36 repo: NewsItemRepository, 

37 application_repo: ApplicationRepository, 

38 owner: Owner, 

39 presenter: Presenter[NewsItemEntity], 

40 ): 

41 """Initialize the use case. 

42 

43 Args: 

44 repo: A repository for updating news items. 

45 application_repo: A repository for getting the application. 

46 owner: The owner of the news item. 

47 presenter: A presenter for a news item. 

48 """ 

49 self._repo = repo 

50 self._application_repo = application_repo 

51 self._owner = owner 

52 self._presenter = presenter 

53 

54 async def execute(self, command: UpdateNewsItemCommand): 

55 """Execute the use case. 

56 

57 Args: 

58 command: The input for this use case. 

59 

60 Raises: 

61 NewsItemNotFoundException: When the news item does not exist. 

62 ApplicationNotFoundException: When the application does not exist. 

63 """ 

64 news_item = await self._repo.get_by_id(NewsItemIdentifier(command.id)) 

65 application = await self._application_repo.get_by_id( 

66 ApplicationIdentifier(command.application) 

67 ) 

68 

69 if command.promotion > 0: 

70 if command.promotion_end_datetime is None: 

71 promotion = Promotion(priority=command.promotion) 

72 else: 

73 promotion = Promotion( 

74 priority=command.promotion, 

75 end_date=Timestamp.create_from_string( 

76 command.promotion_end_datetime 

77 ), 

78 ) 

79 else: 

80 promotion = Promotion() 

81 

82 news_item = replace( 

83 news_item, 

84 enabled=command.enabled, 

85 application=application, 

86 promotion=promotion, 

87 period=Period( 

88 start_date=Timestamp.create_from_string(command.publish_datetime), 

89 end_date=( 

90 Timestamp() 

91 if command.end_datetime 

92 else Timestamp.create_from_string(command.end_datetime) 

93 ), 

94 ), 

95 texts=tuple( 

96 [ 

97 LocaleText( 

98 locale=Locale(text.locale), 

99 format=DocumentFormat(text.format), 

100 title=text.title, 

101 content=text.content, 

102 summary=text.summary, 

103 author=self._owner, 

104 ) 

105 for text in command.texts 

106 ] 

107 ), 

108 remark=command.remark, 

109 traceable_time=news_item.traceable_time.mark_for_update(), 

110 ) 

111 

112 await self._repo.update(news_item) 

113 self._presenter.present(news_item)