Coverage for src/kwai/modules/training/trainings/training_tables.py: 99%
91 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 trainings."""
3from dataclasses import dataclass
4from datetime import datetime, time
6from kwai.core.db.rows import TextRow
7from kwai.core.db.table import Table
8from kwai.core.db.table_row import TableRow
9from kwai.core.domain.value_objects.owner import Owner
10from kwai.core.domain.value_objects.period import Period
11from kwai.core.domain.value_objects.text import LocaleText
12from kwai.core.domain.value_objects.time_period import TimePeriod
13from kwai.core.domain.value_objects.timestamp import Timestamp
14from kwai.core.domain.value_objects.traceable_time import TraceableTime
15from kwai.core.domain.value_objects.weekday import Weekday
16from kwai.modules.training.coaches.coach import CoachEntity
17from kwai.modules.training.teams.team import TeamEntity
18from kwai.modules.training.trainings.training import (
19 TrainingEntity,
20 TrainingIdentifier,
21)
22from kwai.modules.training.trainings.training_definition import (
23 TrainingDefinitionEntity,
24 TrainingDefinitionIdentifier,
25)
26from kwai.modules.training.trainings.value_objects import TrainingCoach
29@dataclass(kw_only=True, frozen=True, slots=True)
30class TrainingTextRow(TextRow):
31 """Represent a row in the training_contents table.
33 Attributes:
34 training_id: The id of the training
35 """
37 training_id: int
39 @classmethod
40 def persist(cls, training: TrainingEntity, content: LocaleText):
41 """Persist a content value object to this table.
43 Args:
44 training: The training that contains the text content.
45 content: The text content of the training.
46 """
47 return TrainingTextRow(
48 training_id=training.id.value,
49 locale=content.locale.value,
50 format=content.format.value,
51 title=content.title,
52 content=content.content,
53 summary=content.summary,
54 user_id=content.author.id.value,
55 created_at=content.traceable_time.created_at.timestamp,
56 updated_at=content.traceable_time.updated_at.timestamp,
57 )
60TrainingContentsTable = Table("training_contents", TrainingTextRow)
63@dataclass(kw_only=True, frozen=True, slots=True)
64class TrainingRow:
65 """Represent a table row of the trainings table.
67 Attributes:
68 id: the id of the training
69 definition_id: the id of the relation training definition
70 season_id: the id of the related season
71 created_at: the timestamp of creation
72 updated_at: the timestamp of the last modification
73 start_date: the timestamp of the start of the training
74 end_date: the timestamp of the end of the training
75 active: is this training active?
76 cancelled: is this training cancelled?
77 location: the location of the training
78 remark: a remark about the training
79 """
81 id: int
82 definition_id: int | None
83 season_id: int | None
84 created_at: datetime
85 updated_at: datetime | None
86 start_date: datetime
87 end_date: datetime
88 active: int
89 cancelled: int
90 location: str | None
91 remark: str | None
93 def create_entity(
94 self,
95 content: list[LocaleText],
96 definition: TrainingDefinitionEntity | None = None,
97 ) -> TrainingEntity:
98 """Create a training entity from the table row.
100 Returns:
101 A training entity.
102 """
103 return TrainingEntity(
104 id_=TrainingIdentifier(self.id),
105 texts=content,
106 definition=definition,
107 period=Period(
108 start_date=Timestamp(self.start_date),
109 end_date=Timestamp(self.end_date),
110 ),
111 active=self.active == 1,
112 cancelled=self.cancelled == 1,
113 location=self.location,
114 remark=self.remark,
115 traceable_time=TraceableTime(
116 created_at=Timestamp(self.created_at),
117 updated_at=Timestamp(self.updated_at),
118 ),
119 )
121 @classmethod
122 def persist(cls, training: TrainingEntity) -> "TrainingRow":
123 """Persist a training.
125 Args:
126 training: The training to persist.
128 Returns:
129 A dataclass containing the table row data.
130 """
131 return TrainingRow(
132 id=training.id.value,
133 definition_id=(
134 None if training.definition is None else training.definition.id.value
135 ),
136 season_id=None,
137 created_at=training.traceable_time.created_at.timestamp,
138 updated_at=training.traceable_time.updated_at.timestamp,
139 start_date=training.period.start_date.timestamp,
140 end_date=training.period.end_date.timestamp,
141 active=1 if training.active else 0,
142 cancelled=1 if training.cancelled else 0,
143 location=training.location,
144 remark=training.remark or "",
145 )
148TrainingsTable = Table("trainings", TrainingRow)
151@dataclass(kw_only=True, frozen=True, slots=True)
152class TrainingDefinitionRow:
153 """Represent a table row of the training definitions table."""
155 id: int
156 name: str
157 description: str
158 season_id: int | None
159 team_id: int | None
160 weekday: int
161 start_time: time
162 end_time: time
163 active: int
164 location: str | None
165 remark: str | None
166 user_id: int
167 created_at: datetime
168 updated_at: datetime | None
169 timezone: str
171 def create_entity(
172 self, team: TeamEntity | None, owner: Owner
173 ) -> TrainingDefinitionEntity:
174 """Create a training definition entity from a table row.
176 Args:
177 team: A team that is associated with the definition.
178 owner: The owner of the training definition.
180 Returns:
181 A training definition entity.
182 """
183 return TrainingDefinitionEntity(
184 id_=TrainingDefinitionIdentifier(self.id),
185 name=self.name,
186 description=self.description,
187 weekday=Weekday(self.weekday),
188 period=TimePeriod(
189 start=self.start_time,
190 end=self.end_time,
191 timezone=self.timezone,
192 ),
193 active=self.active == 1,
194 location=self.location or "",
195 remark=self.remark or "",
196 team=team,
197 owner=owner,
198 traceable_time=TraceableTime(
199 created_at=Timestamp(self.created_at),
200 updated_at=Timestamp(self.updated_at),
201 ),
202 )
204 @classmethod
205 def persist(
206 cls, training_definition: TrainingDefinitionEntity
207 ) -> "TrainingDefinitionRow":
208 """Persist a training definition entity."""
209 return TrainingDefinitionRow(
210 id=training_definition.id.value,
211 name=training_definition.name,
212 description=training_definition.description,
213 season_id=None,
214 weekday=training_definition.weekday.value,
215 start_time=training_definition.period.start,
216 end_time=training_definition.period.end,
217 active=1 if training_definition.active else 0,
218 location=training_definition.location,
219 remark=training_definition.remark,
220 team_id=(
221 None
222 if training_definition.team is None
223 else training_definition.team.id.value
224 ),
225 user_id=training_definition.owner.id.value,
226 created_at=training_definition.traceable_time.created_at.timestamp,
227 updated_at=training_definition.traceable_time.updated_at.timestamp,
228 timezone=training_definition.period.timezone,
229 )
232TrainingDefinitionsTable = Table("training_definitions", TrainingDefinitionRow)
235@dataclass(kw_only=True, frozen=True, slots=True)
236class TrainingCoachRow(TableRow):
237 """Represent a row of the training_coaches table."""
239 __table_name__ = "training_coaches"
241 training_id: int
242 coach_id: int
243 coach_type: int
244 present: int
245 payed: int
246 remark: str | None
247 user_id: int
248 created_at: datetime
249 updated_at: datetime | None
251 def create_coach(self, coach: CoachEntity, owner: Owner) -> TrainingCoach:
252 """Create a TrainingCoach value object."""
253 return TrainingCoach(
254 coach=coach,
255 owner=owner,
256 present=self.present == 1,
257 type=self.coach_type,
258 payed=self.payed == 1,
259 remark="" if self.remark is None else self.remark,
260 )
262 @classmethod
263 def persist(cls, training, training_coach: TrainingCoach) -> "TrainingCoachRow":
264 """Persist a TrainingCoach value object into a table row."""
265 return TrainingCoachRow(
266 training_id=training.id.value,
267 coach_id=training_coach.coach.id.value,
268 coach_type=training_coach.type,
269 present=1 if training_coach.present else 0,
270 payed=1 if training_coach.payed else 0,
271 remark=training_coach.remark,
272 user_id=training_coach.owner.id.value,
273 created_at=training_coach.traceable_time.created_at.timestamp,
274 updated_at=training_coach.traceable_time.updated_at.timestamp,
275 )
278@dataclass(kw_only=True, frozen=True, slots=True)
279class TrainingTeamRow:
280 """Represent a row of the training_teams table."""
282 training_id: int
283 team_id: int
285 @classmethod
286 def persist(cls, training: TrainingEntity, team: TeamEntity) -> "TrainingTeamRow":
287 """Persist a team of a training to a table row."""
288 return TrainingTeamRow(
289 training_id=training.id.value,
290 team_id=team.id.value,
291 )
294TrainingTeamsTable = Table("training_teams", TrainingTeamRow)