Coverage for bc/kwai-bc-training/src/kwai_bc_training/trainings/_tables.py: 98%
57 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 TrainingEntity,
20 TrainingIdentifier,
21)
22from kwai_bc_training.trainings.training_schedule import (
23 TrainingScheduleEntity,
24 TrainingScheduleIdentifier,
25)
26from kwai_bc_training.trainings.value_objects import TrainingCoach
29@dataclass(kw_only=True, frozen=True, slots=True)
30class TrainingTextRow(TextRow, TableRow):
31 """Represent a row in the training_contents table.
33 Attributes:
34 training_id: The id of the training
35 """
37 __table_name__ = "training_contents"
39 training_id: int
41 @classmethod
42 def persist(cls, training: TrainingEntity, content: LocaleText):
43 """Persist a content value object to this table.
45 Args:
46 training: The training that contains the text content.
47 content: The text content of the training.
48 """
49 return TrainingTextRow(
50 training_id=training.id.value,
51 locale=content.locale.value,
52 format=content.format.value,
53 title=content.title,
54 content=content.content,
55 summary=content.summary,
56 user_id=content.author.id.value,
57 created_at=unwrap(content.traceable_time.created_at.timestamp),
58 updated_at=content.traceable_time.updated_at.timestamp,
59 )
62@dataclass(kw_only=True, frozen=True, slots=True)
63class TrainingRow(TableRow):
64 """Represent a table row of the trainings table.
66 Attributes:
67 id: the id of the training
68 training_schedule_id: the id of the related training schema
69 season_id: the id of the related season
70 created_at: the timestamp of creation
71 updated_at: the timestamp of the last modification
72 start_date: the timestamp of the start of the training
73 end_date: the timestamp of the end of the training
74 active: is this training active?
75 cancelled: is this training cancelled?
76 location: the location of the training
77 remark: a remark about the training
78 """
80 __table_name__ = "trainings"
82 id: int
83 training_schedule_id: int | None
84 season_id: int | None
85 created_at: datetime
86 updated_at: datetime | None
87 start_date: datetime
88 end_date: datetime
89 active: int
90 cancelled: int
91 location: str | None
92 remark: str | None
94 def create_entity(
95 self,
96 content: tuple[LocaleText, ...],
97 training_schema: TrainingScheduleEntity | None = None,
98 ) -> TrainingEntity:
99 """Create a training entity from the table row.
101 Returns:
102 A training entity.
103 """
104 return TrainingEntity(
105 id=TrainingIdentifier(self.id),
106 texts=content,
107 schedule=training_schema,
108 period=Period(
109 start_date=Timestamp(self.start_date),
110 end_date=Timestamp(self.end_date),
111 ),
112 active=self.active == 1,
113 cancelled=self.cancelled == 1,
114 location=self.location or "",
115 remark=self.remark or "",
116 traceable_time=TraceableTime(
117 created_at=Timestamp(self.created_at),
118 updated_at=Timestamp(self.updated_at),
119 ),
120 )
122 @classmethod
123 def persist(cls, training: TrainingEntity) -> "TrainingRow":
124 """Persist a training.
126 Args:
127 training: The training to persist.
129 Returns:
130 A dataclass containing the table row data.
131 """
132 return TrainingRow(
133 id=training.id.value,
134 training_schedule_id=(
135 None if training.schedule is None else training.schedule.id.value
136 ),
137 season_id=None,
138 created_at=unwrap(training.traceable_time.created_at.timestamp),
139 updated_at=training.traceable_time.updated_at.timestamp,
140 start_date=unwrap(training.period.start_date.timestamp),
141 end_date=unwrap(training.period.end_date.timestamp),
142 active=1 if training.active else 0,
143 cancelled=1 if training.cancelled else 0,
144 location=training.location,
145 remark=training.remark or "",
146 )
149@dataclass(kw_only=True, frozen=True, slots=True)
150class TrainingScheduleRow(TableRow):
151 """Represent a table row of the training schedule table."""
153 __table_name__ = "training_schedules"
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: timedelta
162 end_time: timedelta
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 ) -> TrainingScheduleEntity:
174 """Create a training schedule entity from a table row.
176 Args:
177 team: A team that is associated with the schedule.
178 owner: The owner of the training schedule.
180 Returns:
181 A training schedule entity.
182 """
183 return TrainingScheduleEntity(
184 id=TrainingScheduleIdentifier(self.id),
185 name=self.name,
186 description=self.description,
187 weekday=Weekday(self.weekday),
188 period=TimePeriod(
189 start=(datetime.min + self.start_time).time(),
190 end=(datetime.min + self.end_time).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_schedule: TrainingScheduleEntity
207 ) -> "TrainingScheduleRow":
208 """Persist a training schedule entity."""
209 start_delta = timedelta(
210 hours=training_schedule.period.start.hour,
211 minutes=training_schedule.period.start.minute,
212 seconds=training_schedule.period.start.second,
213 )
214 end_delta = timedelta(
215 hours=training_schedule.period.end.hour, # type: ignore
216 minutes=training_schedule.period.end.minute, # type: ignore
217 seconds=training_schedule.period.end.second, # type: ignore
218 )
219 return TrainingScheduleRow(
220 id=training_schedule.id.value,
221 name=training_schedule.name,
222 description=training_schedule.description,
223 season_id=None,
224 weekday=training_schedule.weekday.value,
225 start_time=start_delta,
226 end_time=end_delta,
227 active=1 if training_schedule.active else 0,
228 location=training_schedule.location,
229 remark=training_schedule.remark,
230 team_id=(
231 None
232 if training_schedule.team is None
233 else training_schedule.team.id.value
234 ),
235 user_id=training_schedule.owner.id.value,
236 created_at=unwrap(training_schedule.traceable_time.created_at.timestamp),
237 updated_at=training_schedule.traceable_time.updated_at.timestamp,
238 timezone=training_schedule.period.timezone,
239 )
242@dataclass(kw_only=True, frozen=True, slots=True)
243class TrainingScheduleCoachRow(TableRow):
244 """Represent a row of the training_schedule_coaches table."""
246 __table_name__ = "training_schedule_coaches"
248 training_schedule_id: int
249 coach_id: int
252@dataclass(kw_only=True, frozen=True, slots=True)
253class TrainingCoachRow(TableRow):
254 """Represent a row of the training_coaches table."""
256 __table_name__ = "training_coaches"
258 training_id: int
259 coach_id: int
260 coach_type: int
261 present: int
262 payed: int
263 remark: str | None
264 user_id: int
265 created_at: datetime
266 updated_at: datetime | None
268 def create_coach(self, coach: CoachEntity, owner: Owner) -> TrainingCoach:
269 """Create a TrainingCoach value object."""
270 return TrainingCoach(
271 coach=coach,
272 owner=owner,
273 present=self.present == 1,
274 type=self.coach_type,
275 payed=self.payed == 1,
276 remark="" if self.remark is None else self.remark,
277 )
279 @classmethod
280 def persist(cls, training, training_coach: TrainingCoach) -> "TrainingCoachRow":
281 """Persist a TrainingCoach value object into a table row."""
282 return TrainingCoachRow(
283 training_id=training.id.value,
284 coach_id=training_coach.coach.id.value,
285 coach_type=training_coach.type,
286 present=1 if training_coach.present else 0,
287 payed=1 if training_coach.payed else 0,
288 remark=training_coach.remark,
289 user_id=training_coach.owner.id.value,
290 created_at=unwrap(training_coach.traceable_time.created_at.timestamp),
291 updated_at=training_coach.traceable_time.updated_at.timestamp,
292 )
295@dataclass(kw_only=True, frozen=True, slots=True)
296class TrainingTeamRow(TableRow):
297 """Represent a row of the training_teams table."""
299 __table_name__ = "training_teams"
301 training_id: int
302 team_id: int
304 @classmethod
305 def persist(cls, training: TrainingEntity, team: TeamEntity) -> "TrainingTeamRow":
306 """Persist a team of a training to a table row."""
307 return TrainingTeamRow(
308 training_id=training.id.value,
309 team_id=team.id.value,
310 )