Coverage for bc/kwai-bc-portal/src/kwai_bc_portal/news/_tables.py: 100%

1"""Module that defines all dataclasses for the tables containing news items.""" 

2 

3from dataclasses import dataclass 

4from datetime import datetime 

5from typing import Self 

6 

7from kwai_core.db.rows import TextRow 

8from kwai_core.db.table_row import TableRow 

9from kwai_core.domain.value_objects.period import Period 

10from kwai_core.domain.value_objects.text import LocaleText 

11from kwai_core.domain.value_objects.timestamp import Timestamp 

12from kwai_core.domain.value_objects.traceable_time import TraceableTime 

13 

14from kwai_bc_portal.domain.application import ApplicationEntity 

15from kwai_bc_portal.domain.news_item import ( 

16 NewsItemEntity, 

17 NewsItemIdentifier, 

18 Promotion, 

19) 

20 

21 

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

23class NewsItemTextRow(TextRow, TableRow): 

24 """Represent a row in the news_contents table. 

25 

26 Attributes: 

27 news_id: The id of the news item 

28 """ 

29 

30 __table_name__ = "news_contents" 

31 

32 news_id: int 

33 

34 @classmethod 

35 def persist(cls, news_item: NewsItemEntity, text: LocaleText) -> Self: 

36 """Persist a content value object to the table. 

37 

38 Args: 

39 news_item: The news item that contains the content. 

40 text: The text of a news item. 

41 """ 

42 return cls( 

43 news_id=news_item.id.value, 

44 locale=text.locale.value, 

45 format=text.format.value, 

46 title=text.title, 

47 content=text.content, 

48 summary=text.summary, 

49 user_id=text.author.id.value, 

50 created_at=text.traceable_time.created_at.timestamp, # type: ignore 

51 updated_at=text.traceable_time.updated_at.timestamp, 

52 ) 

53 

54 

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

56class NewsItemRow(TableRow): 

57 """Represent a table row of the news items table. 

58 

59 Attributes: 

60 id: the id of the news item 

61 enabled: is the news item enabled? 

62 promotion: the priority to use for the promotion 

63 promotion_end_date: when ends the promotion? 

64 publish_date: time of publication 

65 end_date: end of publication 

66 remark: a remark about the news item 

67 application_id: the link to the application 

68 created_at: the timestamp of creation 

69 updated_at: the timestamp of the last modification 

70 """ 

71 

72 __table_name__ = "news_stories" 

73 

74 id: int 

75 enabled: int 

76 promotion: int 

77 promotion_end_date: datetime | None 

78 publish_date: datetime 

79 end_date: datetime | None 

80 remark: str | None 

81 application_id: int 

82 created_at: datetime 

83 updated_at: datetime | None 

84 

85 def create_entity( 

86 self, application: ApplicationEntity, texts: tuple[LocaleText, ...] 

87 ) -> NewsItemEntity: 

88 """Create a news item entity from a table row.""" 

89 return NewsItemEntity( 

90 id=NewsItemIdentifier(self.id), 

91 enabled=self.enabled == 1, 

92 promotion=Promotion( 

93 priority=self.promotion, 

94 end_date=Timestamp.create_utc(self.promotion_end_date), 

95 ), 

96 period=Period( 

97 start_date=Timestamp.create_utc(self.publish_date), 

98 end_date=Timestamp.create_utc(self.end_date), 

99 ), 

100 application=application, 

101 texts=texts, 

102 remark=self.remark or "", 

103 traceable_time=TraceableTime( 

104 created_at=Timestamp.create_utc(timestamp=self.created_at), 

105 updated_at=Timestamp.create_utc(timestamp=self.updated_at), 

106 ), 

107 ) 

108 

109 @classmethod 

110 def persist(cls, news_item: NewsItemEntity) -> Self: 

111 """Persist an entity to row data. 

112 

113 Args: 

114 news_item: The news item entity to persist. 

115 """ 

116 return cls( 

117 id=news_item.id.value, 

118 enabled=1 if news_item.enabled else 0, 

119 promotion=news_item.promotion.priority, 

120 promotion_end_date=( 

121 None 

122 if news_item.promotion.end_date.empty 

123 else news_item.promotion.end_date.timestamp 

124 ), 

125 publish_date=news_item.period.start_date.timestamp, # type: ignore 

126 end_date=( 

127 None 

128 if news_item.period.endless 

129 else news_item.period.end_date.timestamp 

130 ), 

131 remark=news_item.remark, 

132 application_id=news_item.application.id.value, 

133 created_at=news_item.traceable_time.created_at.timestamp, # type: ignore 

134 updated_at=news_item.traceable_time.updated_at.timestamp, 

135 )