Coverage for src/kwai/modules/portal/news/news_tables.py: 100%
35 statements
« prev ^ index » next coverage.py v7.6.10, created at 2024-01-01 00:00 +0000
1"""Module that defines all dataclasses for the tables containing stories."""
3from dataclasses import dataclass
4from datetime import datetime
6from kwai.core.db.rows import TextRow
7from kwai.core.db.table import Table
8from kwai.core.domain.value_objects.period import Period
9from kwai.core.domain.value_objects.text import LocaleText
10from kwai.core.domain.value_objects.timestamp import Timestamp
11from kwai.core.domain.value_objects.traceable_time import TraceableTime
12from kwai.modules.portal.applications.application import ApplicationEntity
13from kwai.modules.portal.news.news_item import (
14 NewsItemEntity,
15 NewsItemIdentifier,
16 Promotion,
17)
20@dataclass(kw_only=True, frozen=True, slots=True)
21class NewsItemTextRow(TextRow):
22 """Represent a row in the news_contents table.
24 Attributes:
25 news_id: The id of the news item
26 """
28 news_id: int
30 @classmethod
31 def persist(cls, news_item: NewsItemEntity, text: LocaleText):
32 """Persist a content value object to the table.
34 Args:
35 news_item: The news item that contains the content.
36 text: The text of a news item.
37 """
38 return NewsItemTextRow(
39 news_id=news_item.id.value,
40 locale=text.locale.value,
41 format=text.format.value,
42 title=text.title,
43 content=text.content,
44 summary=text.summary,
45 user_id=text.author.id.value,
46 created_at=text.traceable_time.created_at.timestamp,
47 updated_at=text.traceable_time.updated_at.timestamp,
48 )
51NewsItemTextsTable = Table("news_contents", NewsItemTextRow)
54@dataclass(kw_only=True, frozen=True, slots=True)
55class NewsItemRow:
56 """Represent a table row of the news items table.
58 Attributes:
59 id: the id of the news item
60 enabled: is the news item enabled?
61 promotion: the priority to use for the promotion
62 promotion_end_date: when ends the promotion?
63 publish_date: time of publication
64 end_date: end of publication
65 remark: a remark about the news item
66 application_id: the link to the application
67 created_at: the timestamp of creation
68 updated_at: the timestamp of the last modification
69 """
71 id: int
72 enabled: int
73 promotion: int
74 promotion_end_date: datetime | None
75 publish_date: datetime
76 end_date: datetime | None
77 remark: str | None
78 application_id: int
79 created_at: datetime
80 updated_at: datetime | None
82 def create_entity(
83 self, application: ApplicationEntity, texts: list[LocaleText]
84 ) -> NewsItemEntity:
85 """Create a news item entity from a table row."""
86 return NewsItemEntity(
87 id_=NewsItemIdentifier(self.id),
88 enabled=self.enabled == 1,
89 promotion=Promotion(
90 priority=self.promotion,
91 end_date=Timestamp.create_utc(self.promotion_end_date),
92 ),
93 period=Period(
94 start_date=Timestamp.create_utc(self.publish_date),
95 end_date=Timestamp.create_utc(self.end_date),
96 ),
97 application=application,
98 texts=texts,
99 remark=self.remark or "",
100 traceable_time=TraceableTime(
101 created_at=Timestamp.create_utc(timestamp=self.created_at),
102 updated_at=Timestamp.create_utc(timestamp=self.updated_at),
103 ),
104 )
106 @classmethod
107 def persist(cls, news_item: NewsItemEntity) -> "NewsItemRow":
108 """Persist an entity to row data.
110 Args:
111 news_item: The news item entity to persist.
112 """
113 return NewsItemRow(
114 id=news_item.id.value,
115 enabled=1 if news_item.is_enabled else 0,
116 promotion=news_item.promotion.priority,
117 promotion_end_date=news_item.promotion.end_date.timestamp,
118 publish_date=news_item.period.start_date.timestamp,
119 end_date=(
120 None
121 if news_item.period.endless
122 else news_item.period.end_date.timestamp
123 ),
124 remark=news_item.remark,
125 application_id=news_item.application.id.value,
126 created_at=news_item.traceable_time.created_at.timestamp,
127 updated_at=news_item.traceable_time.updated_at.timestamp,
128 )
131NewsItemsTable = Table("news_stories", NewsItemRow)