Coverage for kwai/modules/news/stories/story_tables.py: 100%

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

2from dataclasses import dataclass 

3from datetime import datetime 

4 

5from kwai.core.db.rows import ContentRow 

6from kwai.core.db.table import Table 

7from kwai.core.domain.value_objects.identifier import IntIdentifier 

8from kwai.core.domain.value_objects.local_timestamp import LocalTimestamp 

9from kwai.core.domain.value_objects.period import Period 

10from kwai.core.domain.value_objects.text import LocaleText 

11from kwai.core.domain.value_objects.traceable_time import TraceableTime 

12from kwai.modules.news.stories.story import ( 

13 Application, 

14 Promotion, 

15 StoryEntity, 

16 StoryIdentifier, 

17) 

18 

19 

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

21class StoryContentRow(ContentRow): 

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

23 

24 Attributes: 

25 news_id: The id of the news story 

26 """ 

27 

28 news_id: int 

29 

30 @classmethod 

31 def persist(cls, story: StoryEntity, content: LocaleText): 

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

33 

34 Args: 

35 story: The story that contains the content. 

36 content: The content of a story. 

37 """ 

38 return StoryContentRow( 

39 news_id=story.id.value, 

40 locale=content.locale, 

41 format=content.format, 

42 title=content.title, 

43 content=content.content, 

44 summary=content.summary, 

45 user_id=content.author.id.value, 

46 created_at=content.traceable_time.created_at.timestamp, 

47 updated_at=content.traceable_time.updated_at.timestamp, 

48 ) 

49 

50 

51StoryContentsTable = Table("news_contents", StoryContentRow) 

52 

53 

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

55class ApplicationRow: 

56 """Represent the application data that is associated with a story.""" 

57 

58 id: int 

59 name: str 

60 title: str 

61 

62 def create_application(self) -> Application: 

63 """Create an Application value object from row data.""" 

64 return Application(id=IntIdentifier(self.id), name=self.name, title=self.title) 

65 

66 

67ApplicationsTable = Table("applications", ApplicationRow) 

68 

69 

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

71class StoryRow: 

72 """Represent a table row of the stories table. 

73 

74 Attributes: 

75 id: the id of the story 

76 enabled: is the story enabled? 

77 promotion: the priority to use for the promotion 

78 promotion_end_date: when ends the promotion? 

79 publish_date: time of publication 

80 end_date: end of publication 

81 remark: a remark about the story 

82 application_id: the link to the application 

83 created_at: the timestamp of creation 

84 updated_at: the timestamp of the last modification 

85 """ 

86 

87 id: int 

88 enabled: int 

89 promotion: int 

90 promotion_end_date: datetime | None 

91 publish_date: datetime 

92 end_date: datetime | None 

93 remark: str | None 

94 application_id: int 

95 created_at: datetime 

96 updated_at: datetime | None 

97 

98 def create_entity( 

99 self, application: Application, content: list[LocaleText] 

100 ) -> StoryEntity: 

101 """Create a story entity from a table row.""" 

102 return StoryEntity( 

103 id_=StoryIdentifier(self.id), 

104 enabled=self.enabled == 1, 

105 promotion=Promotion( 

106 priority=self.promotion, 

107 end_date=LocalTimestamp(self.promotion_end_date), 

108 ), 

109 period=Period( 

110 start_date=LocalTimestamp(self.publish_date), 

111 end_date=LocalTimestamp(self.end_date), 

112 ), 

113 application=application, 

114 content=content, 

115 remark=self.remark, 

116 traceable_time=TraceableTime( 

117 created_at=LocalTimestamp(timestamp=self.created_at), 

118 updated_at=LocalTimestamp(timestamp=self.updated_at), 

119 ), 

120 ) 

121 

122 @classmethod 

123 def persist(cls, story: StoryEntity) -> "StoryRow": 

124 """Persist an entity to row data. 

125 

126 Args: 

127 story: The story to persist. 

128 """ 

129 return StoryRow( 

130 id=story.id.value, 

131 enabled=1 if story.is_enabled else 0, 

132 promotion=story.promotion.priority, 

133 promotion_end_date=story.promotion.end_date.timestamp, 

134 publish_date=story.period.start_date.timestamp, 

135 end_date=None if story.period.endless else story.period.end_date.timestamp, 

136 remark=story.remark, 

137 application_id=story.application.id.value, 

138 created_at=story.traceable_time.created_at.timestamp, 

139 updated_at=story.traceable_time.updated_at.timestamp, 

140 ) 

141 

142 

143StoriesTable = Table("news_stories", StoryRow)