Coverage for src/kwai/modules/club/repositories/_tables.py: 100%
127 statements
« prev ^ index » next coverage.py v7.6.10, 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.core.db.table_row import TableRow
8from kwai.core.domain.value_objects.date import Date
9from kwai.core.domain.value_objects.email_address import EmailAddress
10from kwai.core.domain.value_objects.name import Name
11from kwai.core.domain.value_objects.timestamp import Timestamp
12from kwai.core.domain.value_objects.traceable_time import TraceableTime
13from kwai.core.domain.value_objects.unique_id import UniqueId
14from kwai.modules.club.domain.coach import CoachEntity
15from kwai.modules.club.domain.contact import ContactEntity, ContactIdentifier
16from kwai.modules.club.domain.country import CountryEntity, CountryIdentifier
17from kwai.modules.club.domain.file_upload import FileUploadEntity
18from kwai.modules.club.domain.member import MemberEntity, MemberIdentifier
19from kwai.modules.club.domain.person import PersonEntity, PersonIdentifier
20from kwai.modules.club.domain.value_objects import (
21 Address,
22 Birthdate,
23 Gender,
24 License,
25)
28@dataclass(kw_only=True, frozen=True, slots=True)
29class CountryRow(TableRow):
30 """Represent a row of the countries table.
32 Attributes:
33 id: The id of the country.
34 iso_2: The ISO 2 code of the country.
35 iso_3: The ISO 3 code of the country.
36 """
38 __table_name__ = "countries"
40 id: int | None = None
41 iso_2: str
42 iso_3: str
43 name: str
44 created_at: datetime
45 updated_at: datetime | None
47 def create_country(self) -> CountryEntity:
48 """Create a Country value object from the row.
50 Returns:
51 A country value object.
52 """
53 return CountryEntity(
54 id_=CountryIdentifier(self.id),
55 iso_2=self.iso_2,
56 iso_3=self.iso_3,
57 name=self.name,
58 )
60 @classmethod
61 def persist(cls, country: CountryEntity):
62 """Persist a country to this table.
64 Args:
65 country: The country to persist.
66 """
67 return cls(
68 iso_2=country.iso_2,
69 iso_3=country.iso_3,
70 name=country.name,
71 created_at=datetime.now(UTC),
72 updated_at=None,
73 )
76@dataclass(kw_only=True, frozen=True, slots=True)
77class FileUploadRow(TableRow):
78 """Represents a row of the imports table.
80 Attributes:
81 id: The id of the fileupload.
82 filename: The name of the uploaded file.
83 user_id: The id of the user that uploaded the file.
84 created_at: The timestamp of creation.
85 updated_at: The timestamp of modification.
86 """
88 __table_name__ = "imports"
90 id: int
91 uuid: str
92 filename: str
93 remark: str
94 preview: int
95 user_id: int
96 created_at: datetime
97 updated_at: datetime | None
99 @classmethod
100 def persist(cls, file_upload: FileUploadEntity) -> Self:
101 """Persist a file upload entity to this table.
103 Args:
104 file_upload: The entity to persist.
105 """
106 return cls(
107 id=file_upload.id.value,
108 uuid=str(file_upload.uuid),
109 filename=file_upload.filename,
110 remark=file_upload.remark,
111 preview=1 if file_upload.preview else 0,
112 user_id=file_upload.owner.id.value,
113 created_at=file_upload.traceable_time.created_at.timestamp, # type: ignore[arg-type]
114 updated_at=file_upload.traceable_time.updated_at.timestamp,
115 )
118@dataclass(kw_only=True, frozen=True, slots=True)
119class ContactRow(TableRow):
120 """Represents a row of the contacts table."""
122 __table_name__ = "contacts"
124 id: int
125 email: str
126 tel: str
127 mobile: str
128 address: str
129 postal_code: str
130 city: str
131 county: str | None
132 country_id: int
133 remark: str | None
134 created_at: datetime
135 updated_at: datetime | None
137 def create_entity(self, country: CountryEntity) -> ContactEntity:
138 """Create a contact entity from a table row."""
139 emails = [EmailAddress(email) for email in self.email.split(";")]
140 return ContactEntity(
141 id_=ContactIdentifier(self.id),
142 emails=emails,
143 tel=self.tel,
144 mobile=self.mobile,
145 remark=self.remark or "",
146 address=Address(
147 address=self.address,
148 postal_code=self.postal_code,
149 city=self.city,
150 county=self.county or "",
151 country=country,
152 ),
153 )
155 @classmethod
156 def persist(cls, contact: ContactEntity) -> Self:
157 """Create a row from a contact entity."""
158 return cls(
159 id=contact.id.value,
160 email=";".join([str(email) for email in contact.emails]),
161 tel=contact.tel,
162 mobile=contact.mobile,
163 address=contact.address.address,
164 postal_code=contact.address.postal_code,
165 city=contact.address.city,
166 county=contact.address.county,
167 country_id=contact.address.country.id.value,
168 remark=contact.remark,
169 created_at=contact.traceable_time.created_at.timestamp, # type: ignore[arg-type]
170 updated_at=contact.traceable_time.updated_at.timestamp,
171 )
174@dataclass(kw_only=True, frozen=True, slots=True)
175class PersonRow(TableRow):
176 """Represents a row of the persons table."""
178 __table_name__ = "persons"
180 id: int
181 lastname: str
182 firstname: str
183 gender: int
184 birthdate: date
185 remark: str | None
186 user_id: int | None
187 contact_id: int
188 nationality_id: int
189 created_at: datetime
190 updated_at: datetime | None
192 def create_entity(
193 self, nationality: CountryEntity, contact: ContactEntity
194 ) -> PersonEntity:
195 """Create a person entity from a table row."""
196 return PersonEntity(
197 id_=PersonIdentifier(self.id),
198 name=Name(last_name=self.lastname, first_name=self.firstname),
199 gender=Gender(self.gender),
200 birthdate=Birthdate(Date.create_from_date(self.birthdate)),
201 remark=self.remark or "",
202 contact=contact,
203 traceable_time=TraceableTime(
204 created_at=Timestamp.create_utc(self.created_at),
205 updated_at=Timestamp.create_utc(self.updated_at),
206 ),
207 nationality=nationality,
208 )
210 @classmethod
211 def persist(cls, person: PersonEntity) -> Self:
212 """Create a row from a person entity."""
213 return cls(
214 id=person.id.value,
215 lastname=person.name.last_name or "",
216 firstname=person.name.first_name or "",
217 gender=person.gender.value,
218 birthdate=person.birthdate.date.date,
219 remark=person.remark,
220 user_id=None,
221 contact_id=person.contact.id.value,
222 nationality_id=person.nationality.id.value,
223 created_at=person.traceable_time.created_at.timestamp, # type: ignore[arg-type]
224 updated_at=person.traceable_time.updated_at.timestamp,
225 )
228@dataclass(kw_only=True, frozen=True, slots=True)
229class MemberRow(TableRow):
230 """Represents a row of the members table."""
232 __table_name__ = "judo_members"
234 id: int
235 uuid: str
236 license: str
237 license_end_date: date
238 person_id: int
239 remark: str | None
240 competition: int
241 created_at: datetime
242 updated_at: datetime | None
243 active: int
245 def create_entity(self, person: PersonEntity) -> MemberEntity:
246 """Create a member entity of a table row."""
247 return MemberEntity(
248 id_=MemberIdentifier(self.id),
249 uuid=UniqueId.create_from_string(self.uuid),
250 license=License(
251 number=self.license,
252 end_date=Date.create_from_date(self.license_end_date),
253 ),
254 remark=self.remark or "",
255 competition=self.competition == 1,
256 active=self.active == 1,
257 person=person,
258 traceable_time=TraceableTime(
259 created_at=Timestamp.create_utc(self.created_at),
260 updated_at=Timestamp.create_utc(self.updated_at),
261 ),
262 )
264 @classmethod
265 def persist(cls, member: MemberEntity) -> Self:
266 """Create a row from the member entity."""
267 return cls(
268 id=member.id.value,
269 uuid=str(member.uuid),
270 license=member.license.number,
271 license_end_date=member.license.end_date.date,
272 person_id=member.person.id.value,
273 remark=member.remark,
274 competition=1 if member.is_competitive else 0,
275 active=1 if member.is_active else 0,
276 created_at=member.traceable_time.created_at.timestamp, # type: ignore[arg-type]
277 updated_at=member.traceable_time.updated_at.timestamp,
278 )
281@dataclass(kw_only=True, frozen=True, slots=True)
282class MemberUploadRow(TableRow):
283 """Represents a row of the judo member imports table."""
285 __table_name__ = "judo_member_imports"
287 member_id: int
288 import_id: int
289 created_at: datetime
291 @classmethod
292 def persist(cls, upload: FileUploadEntity, member: MemberEntity) -> Self:
293 return cls(
294 member_id=member.id.value,
295 import_id=upload.id.value,
296 created_at=datetime.now(UTC),
297 )
300@dataclass(kw_only=True, frozen=True, slots=True)
301class CoachRow(TableRow):
302 """Represents a row of the coach table."""
304 __table_name__ = "coaches"
306 id: int
307 member_id: int
308 description: str
309 diploma: str
310 active: int
311 remark: str
312 user_id: int | None
313 created_at: datetime
314 updated_at: datetime | None
316 @classmethod
317 def persist(cls, coach: CoachEntity) -> Self:
318 return cls(
319 id=coach.id.value,
320 member_id=coach.member.id.value,
321 description=coach.description,
322 diploma=coach.diploma,
323 active=1 if coach.is_active else 0,
324 remark=coach.remark,
325 user_id=None if coach.user is None else coach.user.id.value,
326 created_at=coach.traceable_time.created_at.timestamp, # type: ignore[arg-type]
327 updated_at=coach.traceable_time.updated_at.timestamp,
328 )