Coverage for bc/kwai-bc-club/src/kwai_bc_club/repositories/_tables.py: 99%
78 statements
« prev ^ index » next coverage.py v7.11.0, created at 2024-01-01 00:00 +0000
1"""Module that defines all tables related to members."""
3from dataclasses import dataclass
4from datetime import UTC, date, datetime
5from typing import Self
7from kwai_bc_training.coaches.coach import CoachIdentifier
8from kwai_core.db.table_row import TableRow
9from kwai_core.domain.value_objects.date import Date
10from kwai_core.domain.value_objects.email_address import EmailAddress
11from kwai_core.domain.value_objects.name import Name
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.unique_id import UniqueId
16from kwai_bc_club.domain.club_coach import ClubCoachEntity
17from kwai_bc_club.domain.contact import ContactEntity, ContactIdentifier
18from kwai_bc_club.domain.country import CountryEntity, CountryIdentifier
19from kwai_bc_club.domain.file_upload import FileUploadEntity
20from kwai_bc_club.domain.member import MemberEntity, MemberIdentifier
21from kwai_bc_club.domain.person import PersonEntity, PersonIdentifier
22from kwai_bc_club.domain.value_objects import (
23 Address,
24 Birthdate,
25 Gender,
26 License,
27 User,
28)
31@dataclass(kw_only=True, frozen=True, slots=True)
32class CountryRow(TableRow):
33 """Represent a row of the countries table.
35 Attributes:
36 id: The id of the country.
37 iso_2: The ISO 2 code of the country.
38 iso_3: The ISO 3 code of the country.
39 """
41 __table_name__ = "countries"
43 id: int
44 iso_2: str
45 iso_3: str
46 name: str
47 created_at: datetime
48 updated_at: datetime | None
50 def create_country(self) -> CountryEntity:
51 """Create a Country value object from the row.
53 Returns:
54 A country value object.
55 """
56 return CountryEntity(
57 id=CountryIdentifier(self.id),
58 iso_2=self.iso_2,
59 iso_3=self.iso_3,
60 name=self.name,
61 )
63 @classmethod
64 def persist(cls, country: CountryEntity):
65 """Persist a country to this table.
67 Args:
68 country: The country to persist.
69 """
70 return cls(
71 id=country.id.value,
72 iso_2=country.iso_2,
73 iso_3=country.iso_3,
74 name=country.name,
75 created_at=datetime.now(UTC),
76 updated_at=None,
77 )
80@dataclass(kw_only=True, frozen=True, slots=True)
81class FileUploadRow(TableRow):
82 """Represents a row of the imports table.
84 Attributes:
85 id: The id of the fileupload.
86 filename: The name of the uploaded file.
87 user_id: The id of the user that uploaded the file.
88 created_at: The timestamp of creation.
89 updated_at: The timestamp of modification.
90 """
92 __table_name__ = "imports"
94 id: int
95 uuid: str
96 filename: str
97 remark: str
98 preview: int
99 user_id: int
100 created_at: datetime
101 updated_at: datetime | None
103 @classmethod
104 def persist(cls, file_upload: FileUploadEntity) -> Self:
105 """Persist a file upload entity to this table.
107 Args:
108 file_upload: The entity to persist.
109 """
110 return cls(
111 id=file_upload.id.value,
112 uuid=str(file_upload.uuid),
113 filename=file_upload.filename,
114 remark=file_upload.remark,
115 preview=1 if file_upload.preview else 0,
116 user_id=file_upload.owner.id.value,
117 created_at=file_upload.traceable_time.created_at.timestamp, # type: ignore[arg-type]
118 updated_at=file_upload.traceable_time.updated_at.timestamp,
119 )
122@dataclass(kw_only=True, frozen=True, slots=True)
123class ContactRow(TableRow):
124 """Represents a row of the contacts table."""
126 __table_name__ = "contacts"
128 id: int
129 email: str
130 tel: str
131 mobile: str
132 address: str
133 postal_code: str
134 city: str
135 county: str | None
136 country_id: int
137 remark: str | None
138 created_at: datetime
139 updated_at: datetime | None
141 def create_entity(self, country: CountryEntity) -> ContactEntity:
142 """Create a contact entity from a table row."""
143 emails = tuple([EmailAddress(email) for email in self.email.split(";")])
144 return ContactEntity(
145 id=ContactIdentifier(self.id),
146 emails=emails,
147 tel=self.tel,
148 mobile=self.mobile,
149 remark=self.remark or "",
150 address=Address(
151 address=self.address,
152 postal_code=self.postal_code,
153 city=self.city,
154 county=self.county or "",
155 country=country,
156 ),
157 )
159 @classmethod
160 def persist(cls, contact: ContactEntity) -> Self:
161 """Create a row from a contact entity."""
162 return cls(
163 id=contact.id.value,
164 email=";".join([str(email) for email in contact.emails]),
165 tel=contact.tel,
166 mobile=contact.mobile,
167 address=contact.address.address,
168 postal_code=contact.address.postal_code,
169 city=contact.address.city,
170 county=contact.address.county,
171 country_id=contact.address.country.id.value,
172 remark=contact.remark,
173 created_at=contact.traceable_time.created_at.timestamp, # type: ignore[arg-type]
174 updated_at=contact.traceable_time.updated_at.timestamp,
175 )
178@dataclass(kw_only=True, frozen=True, slots=True)
179class PersonRow(TableRow):
180 """Represents a row of the persons table."""
182 __table_name__ = "persons"
184 id: int
185 lastname: str
186 firstname: str
187 gender: int
188 birthdate: date
189 remark: str | None
190 user_id: int | None
191 contact_id: int
192 nationality_id: int
193 created_at: datetime
194 updated_at: datetime | None
196 def create_entity(
197 self, nationality: CountryEntity, contact: ContactEntity
198 ) -> PersonEntity:
199 """Create a person entity from a table row."""
200 return PersonEntity(
201 id=PersonIdentifier(self.id),
202 name=Name(last_name=self.lastname, first_name=self.firstname),
203 gender=Gender(self.gender),
204 birthdate=Birthdate(Date.create_from_date(self.birthdate)),
205 remark=self.remark or "",
206 contact=contact,
207 traceable_time=TraceableTime(
208 created_at=Timestamp.create_utc(self.created_at),
209 updated_at=Timestamp.create_utc(self.updated_at),
210 ),
211 nationality=nationality,
212 )
214 @classmethod
215 def persist(cls, person: PersonEntity) -> Self:
216 """Create a row from a person entity."""
217 return cls(
218 id=person.id.value,
219 lastname=person.name.last_name or "",
220 firstname=person.name.first_name or "",
221 gender=person.gender.value,
222 birthdate=person.birthdate.date.date,
223 remark=person.remark,
224 user_id=None,
225 contact_id=person.contact.id.value,
226 nationality_id=person.nationality.id.value,
227 created_at=person.traceable_time.created_at.timestamp, # type: ignore[arg-type]
228 updated_at=person.traceable_time.updated_at.timestamp,
229 )
232@dataclass(kw_only=True, frozen=True, slots=True)
233class MemberRow(TableRow):
234 """Represents a row of the members table."""
236 __table_name__ = "judo_members"
238 id: int
239 uuid: str
240 license: str
241 license_end_date: date
242 person_id: int
243 remark: str | None
244 competition: int
245 created_at: datetime
246 updated_at: datetime | None
247 active: int
248 user_id: int | None
250 def create_entity(self, person: PersonEntity, user: User | None) -> MemberEntity:
251 """Create a member entity of a table row."""
252 return MemberEntity(
253 id=MemberIdentifier(self.id),
254 uuid=UniqueId.create_from_string(self.uuid),
255 license=License(
256 number=self.license,
257 end_date=Date.create_from_date(self.license_end_date),
258 ),
259 remark=self.remark or "",
260 competition=self.competition == 1,
261 active=self.active == 1,
262 person=person,
263 traceable_time=TraceableTime(
264 created_at=Timestamp.create_utc(self.created_at),
265 updated_at=Timestamp.create_utc(self.updated_at),
266 ),
267 user=user,
268 )
270 @classmethod
271 def persist(cls, member: MemberEntity) -> Self:
272 """Create a row from the member entity."""
273 return cls(
274 id=member.id.value,
275 uuid=str(member.uuid),
276 license=member.license.number,
277 license_end_date=member.license.end_date.date,
278 person_id=member.person.id.value,
279 remark=member.remark,
280 competition=1 if member.competition else 0,
281 active=1 if member.active else 0,
282 created_at=member.traceable_time.created_at.timestamp, # type: ignore[arg-type]
283 updated_at=member.traceable_time.updated_at.timestamp,
284 user_id=None if member.user is None else member.user.id,
285 )
288@dataclass(kw_only=True, frozen=True, slots=True)
289class UserRow(TableRow):
290 """Represents a row of the users table."""
292 __table_name__ = "users"
294 id: int | None
295 uuid: str | None
297 def create_user(self) -> User | None:
298 """Create a User value object from a table row."""
299 if self.id is None or self.uuid is None:
300 return None
301 return User(id=self.id, uuid=UniqueId.create_from_string(self.uuid))
304@dataclass(kw_only=True, frozen=True, slots=True)
305class MemberUploadRow(TableRow):
306 """Represents a row of the judo member imports table."""
308 __table_name__ = "judo_member_imports"
310 member_id: int
311 import_id: int
312 created_at: datetime
314 @classmethod
315 def persist(cls, upload: FileUploadEntity, member: MemberEntity) -> Self:
316 return cls(
317 member_id=member.id.value,
318 import_id=upload.id.value,
319 created_at=datetime.now(UTC),
320 )
323@dataclass(kw_only=True, frozen=True, slots=True)
324class CoachRow(TableRow):
325 """Represents a row of the coach table."""
327 __table_name__ = "coaches"
329 id: int
330 member_id: int
331 description: str | None
332 diploma: str | None
333 active: int
334 remark: str | None
335 created_at: datetime
336 updated_at: datetime | None
338 @classmethod
339 def persist(cls, coach: ClubCoachEntity) -> Self:
340 return cls(
341 id=coach.id.value,
342 member_id=coach.member.id.value,
343 description=coach.description,
344 diploma=coach.diploma,
345 active=1 if coach.active else 0,
346 remark=coach.remark,
347 created_at=coach.traceable_time.created_at.timestamp, # type: ignore[arg-type]
348 updated_at=coach.traceable_time.updated_at.timestamp,
349 )
351 def create_entity(self, member: MemberEntity) -> ClubCoachEntity:
352 return ClubCoachEntity(
353 id=CoachIdentifier(self.id),
354 remark=self.remark or "",
355 description=self.description or "",
356 diploma=self.diploma or "",
357 active=self.active == 1,
358 traceable_time=TraceableTime(
359 created_at=Timestamp.create_utc(self.created_at),
360 updated_at=Timestamp.create_utc(self.updated_at),
361 ),
362 member=member,
363 )