Coverage for bc/kwai-bc-training/src/kwai_bc_training/trainings/_tables.py: 98%
56 statements
« prev ^ index » next coverage.py v7.11.0, 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, timedelta
6from kwai_core.db.rows import TextRow
7from kwai_core.db.table_row import TableRow, unwrap
8from kwai_core.domain.value_objects.owner import Owner
9from kwai_core.domain.value_objects.period import Period
10from kwai_core.domain.value_objects.text import LocaleText
11from kwai_core.domain.value_objects.time_period import TimePeriod
12from kwai_core.domain.value_objects.timestamp import Timestamp
13from kwai_core.domain.value_objects.traceable_time import TraceableTime
14from kwai_core.domain.value_objects.weekday import Weekday
16from kwai_bc_training.coaches.coach import CoachEntity
17from kwai_bc_training.teams.team import TeamEntity
18from kwai_bc_training.trainings.training import (
19 TrainingCoachEntity,
20 TrainingCoachIdentifier,
21 TrainingEntity,
22 TrainingIdentifier,
23)
24from kwai_bc_training.trainings.training_schedule import (
25 TrainingScheduleEntity,
26 TrainingScheduleIdentifier,
27)
30@dataclass(kw_only=True, frozen=True, slots=True)
31class TrainingTextRow(TextRow, TableRow):
32 """Represent a row in the training_contents table.
34 Attributes:
35 training_id: The id of the training
36 """
38 __table_name__ = "training_contents"
40 training_id: int
42 @classmethod
43 def persist(cls, training: TrainingEntity, content: LocaleText):
44 """Persist a content value object to this table.
46 Args:
47 training: The training that contains the text content.
48 content: The text content of the training.
49 """
50 return TrainingTextRow(
51 training_id=training.id.value,
52 locale=content.locale.value,
53 format=content.format.value,
54 title=content.title,
55 content=content.content,
56 summary=content.summary,
57 user_id=content.author.id.value,
58 created_at=unwrap(content.traceable_time.created_at.timestamp),
59 updated_at=content.traceable_time.updated_at.timestamp,
60 )
63@dataclass(kw_only=True, frozen=True, slots=True)
64class TrainingRow(TableRow):
65 """Represent a table row of the trainings table.
67 Attributes:
68 id: the id of the training
69 training_schedule_id: the id of the related training schema
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 __table_name__ = "trainings"
83 id: int
84 training_schedule_id: int | None
85 season_id: int | None
86 created_at: datetime
87 updated_at: datetime | None
88 start_date: datetime
89 end_date: datetime
90 active: int
91 cancelled: int
92 location: str | None
93 remark: str | None
95 def create_entity(
96 self,
97 content: tuple[LocaleText, ...],
98 training_schema: TrainingScheduleEntity | None = None,
99 ) -> TrainingEntity:
100 """Create a training entity from the table row.
102 Returns:
103 A training entity.
104 """
105 return TrainingEntity(
106 id=TrainingIdentifier(self.id),
107 texts=content,
108 schedule=training_schema,
109 period=Period(
110 start_date=Timestamp(self.start_date),
111 end_date=Timestamp(self.end_date),
112 ),
113 active=self.active == 1,
114 cancelled=self.cancelled == 1,
115 location=self.location or "",
116 remark=self.remark or "",
117 traceable_time=TraceableTime(
118 created_at=Timestamp(self.created_at),
119 updated_at=Timestamp(self.updated_at),
120 ),
121 )
123 @classmethod
124 def persist(cls, training: TrainingEntity) -> "TrainingRow":
125 """Persist a training.
127 Args:
128 training: The training to persist.
130 Returns:
131 A dataclass containing the table row data.
132 """
133 return TrainingRow(
134 id=training.id.value,
135 training_schedule_id=(
136 None if training.schedule is None else training.schedule.id.value
137 ),
138 season_id=None,
139 created_at=unwrap(training.traceable_time.created_at.timestamp),
140 updated_at=training.traceable_time.updated_at.timestamp,
141 start_date=unwrap(training.period.start_date.timestamp),
142 end_date=unwrap(training.period.end_date.timestamp),
143 active=1 if training.active else 0,
144 cancelled=1 if training.cancelled else 0,
145 location=training.location,
146 remark=training.remark or "",
147 )
150@dataclass(kw_only=True, frozen=True, slots=True)
151class TrainingScheduleRow(TableRow):
152 """Represent a table row of the training schedule table."""
154 __table_name__ = "training_schedules"
156 id: int
157 name: str
158 description: str
159 season_id: int | None
160 team_id: int | None
161 weekday: int
162 start_time: timedelta
163 end_time: timedelta
164 active: int
165 location: str | None
166 remark: str | None
167 user_id: int
168 created_at: datetime
169 updated_at: datetime | None
170 timezone: str
172 def create_entity(
173 self, team: TeamEntity | None, owner: Owner
174 ) -> TrainingScheduleEntity:
175 """Create a training schedule entity from a table row.
177 Args:
178 team: A team that is associated with the schedule.
179 owner: The owner of the training schedule.
181 Returns:
182 A training schedule entity.
183 """
184 return TrainingScheduleEntity(
185 id=TrainingScheduleIdentifier(self.id),
186 name=self.name,
187 description=self.description,
188 weekday=Weekday(self.weekday),
189 period=TimePeriod(
190 start=(datetime.min + self.start_time).time(),
191 end=(datetime.min + self.end_time).time(),
192 timezone=self.timezone,
193 ),
194 active=self.active == 1,
195 location=self.location or "",
196 remark=self.remark or "",
197 team=team,
198 owner=owner,
199 traceable_time=TraceableTime(
200 created_at=Timestamp(self.created_at),
201 updated_at=Timestamp(self.updated_at),
202 ),
203 )
205 @classmethod
206 def persist(
207 cls, training_schedule: TrainingScheduleEntity
208 ) -> "TrainingScheduleRow":
209 """Persist a training schedule entity."""
210 start_delta = timedelta(
211 hours=training_schedule.period.start.hour,
212 minutes=training_schedule.period.start.minute,
213 seconds=training_schedule.period.start.second,
214 )
215 end_delta = timedelta(
216 hours=training_schedule.period.end.hour, # type: ignore
217 minutes=training_schedule.period.end.minute, # type: ignore
218 seconds=training_schedule.period.end.second, # type: ignore
219 )
220 return TrainingScheduleRow(
221 id=training_schedule.id.value,
222 name=training_schedule.name,
223 description=training_schedule.description,
224 season_id=None,
225 weekday=training_schedule.weekday.value,
226 start_time=start_delta,
227 end_time=end_delta,
228 active=1 if training_schedule.active else 0,
229 location=training_schedule.location,
230 remark=training_schedule.remark,
231 team_id=(
232 None
233 if training_schedule.team is None
234 else training_schedule.team.id.value
235 ),
236 user_id=training_schedule.owner.id.value,
237 created_at=unwrap(training_schedule.traceable_time.created_at.timestamp),
238 updated_at=training_schedule.traceable_time.updated_at.timestamp,
239 timezone=training_schedule.period.timezone,
240 )
243@dataclass(kw_only=True, frozen=True, slots=True)
244class TrainingScheduleCoachRow(TableRow):
245 """Represent a row of the training_schedule_coaches table."""
247 __table_name__ = "training_schedule_coaches"
249 training_schedule_id: int
250 coach_id: int
253@dataclass(kw_only=True, frozen=True, slots=True)
254class TrainingCoachRow(TableRow):
255 """Represent a row of the training_coaches table."""
257 __table_name__ = "training_coaches"
259 id: int
260 training_id: int
261 coach_id: int
262 coach_type: int
263 present: int
264 payed: int
265 remark: str | None
266 user_id: int
267 created_at: datetime
268 updated_at: datetime | None
270 def create_coach(self, coach: CoachEntity, owner: Owner) -> TrainingCoachEntity:
271 """Create a TrainingCoach entity."""
272 return TrainingCoachEntity(
273 id=TrainingCoachIdentifier(self.id),
274 coach=coach,
275 owner=owner,
276 present=self.present == 1,
277 type=self.coach_type,
278 payed=self.payed == 1,
279 remark="" if self.remark is None else self.remark,
280 )
282 @classmethod
283 def persist(
284 cls, training, training_coach: TrainingCoachEntity
285 ) -> "TrainingCoachRow":
286 """Persist a TrainingCoach value object into a table row."""
287 return TrainingCoachRow(
288 id=training_coach.id.value,
289 training_id=training.id.value,
290 coach_id=training_coach.coach.id.value,
291 coach_type=training_coach.type,
292 present=1 if training_coach.present else 0,
293 payed=1 if training_coach.payed else 0,
294 remark=training_coach.remark,
295 user_id=training_coach.owner.id.value,
296 created_at=unwrap(training_coach.traceable_time.created_at.timestamp),
297 updated_at=training_coach.traceable_time.updated_at.timestamp,
298 )
301@dataclass(kw_only=True, frozen=True, slots=True)
302class TrainingTeamRow(TableRow):
303 """Represent a row of the training_teams table."""
305 __table_name__ = "training_teams"
307 training_id: int
308 team_id: int
310 @classmethod
311 def persist(cls, training: TrainingEntity, team: TeamEntity) -> "TrainingTeamRow":
312 """Persist a team of a training to a table row."""
313 return TrainingTeamRow(
314 training_id=training.id.value,
315 team_id=team.id.value,
316 )