Coverage for bc/kwai-bc-teams/src/kwai_bc_teams/repositories/_tables.py: 97%

1from dataclasses import dataclass 

2from datetime import date, datetime 

3from typing import Self 

4 

5from kwai_bc_club.domain.country import CountryEntity, CountryIdentifier 

6from kwai_core.db.table_row import TableRow 

7from kwai_core.domain.value_objects.timestamp import Timestamp 

8from kwai_core.domain.value_objects.traceable_time import TraceableTime 

9from kwai_core.domain.value_objects.unique_id import UniqueId 

10 

11from kwai_bc_teams.domain.team import TeamEntity, TeamIdentifier, TeamMembers 

12from kwai_bc_teams.domain.team_member import MemberEntity, TeamMember 

13 

14 

15@dataclass(kw_only=True, frozen=True, slots=True) 

16class TeamRow(TableRow): 

17 """Represents a row of the teams table.""" 

18 

19 __table_name__ = "teams" 

20 

21 id: int 

22 name: str 

23 season_id: int | None 

24 team_category_id: int | None 

25 active: int 

26 remark: str | None 

27 created_at: datetime 

28 updated_at: datetime | None 

29 

30 def create_entity(self, team_members: dict[UniqueId, TeamMember]) -> TeamEntity: 

31 return TeamEntity( 

32 id=TeamIdentifier(self.id), 

33 name=self.name, 

34 active=self.active == 1, 

35 remark=self.remark or "", 

36 members=TeamMembers(team_members), 

37 traceable_time=TraceableTime( 

38 created_at=Timestamp.create_utc(self.created_at), 

39 updated_at=Timestamp.create_utc(self.updated_at), 

40 ), 

41 ) 

42 

43 @classmethod 

44 def persist(cls, team: TeamEntity) -> Self: 

45 return cls( 

46 id=team.id.value, 

47 name=team.name, 

48 season_id=None, 

49 team_category_id=None, 

50 active=1 if team.active else 0, 

51 remark=team.remark, 

52 created_at=team.traceable_time.created_at.timestamp, # type: ignore[arg-type] 

53 updated_at=team.traceable_time.updated_at.timestamp, 

54 ) 

55 

56 

57@dataclass(kw_only=True, frozen=True, slots=True) 

58class MemberRow(TableRow): 

59 """Represents a row of the members table used for a team member.""" 

60 

61 __table_name__ = "judo_members" 

62 

63 id: int | None 

64 uuid: str | None 

65 license: str | None 

66 license_end_date: date | None 

67 person_id: int | None 

68 active: int | None 

69 

70 

71@dataclass(kw_only=True, frozen=True, slots=True) 

72class MemberPersonRow(TableRow): 

73 """Represents a row of the persons table used for a team member.""" 

74 

75 __table_name__ = "persons" 

76 

77 id: int | None 

78 firstname: str | None 

79 lastname: str | None 

80 gender: int | None 

81 birthdate: date | None 

82 nationality_id: int | None 

83 

84 

85@dataclass(kw_only=True, frozen=True, slots=True) 

86class TeamMemberRow(TableRow): 

87 """Represents a row of the team members table.""" 

88 

89 __table_name__ = "team_members" 

90 

91 team_id: int | None 

92 member_id: int | None 

93 active: int | None 

94 created_at: datetime | None 

95 updated_at: datetime | None 

96 

97 def create_team_member(self, member: MemberEntity): 

98 return TeamMember( 

99 active=True if self.active == 1 else False, 

100 member=member, 

101 traceable_time=TraceableTime( 

102 created_at=Timestamp.create_utc(self.created_at), 

103 updated_at=Timestamp.create_utc(self.updated_at), 

104 ), 

105 ) 

106 

107 @classmethod 

108 def persist(cls, team: TeamEntity, team_member: TeamMember) -> Self: 

109 """Persist a team member to the table row.""" 

110 return cls( 

111 team_id=team.id.value, 

112 member_id=team_member.member.id.value, 

113 active=1 if team_member.active else 0, 

114 created_at=team_member.traceable_time.created_at.timestamp, # type: ignore[arg-type] 

115 updated_at=team_member.traceable_time.updated_at.timestamp, 

116 ) 

117 

118 

119@dataclass(kw_only=True, frozen=True, slots=True) 

120class CountryRow(TableRow): 

121 """Represent a row of the countries table. 

122 

123 Attributes: 

124 id: The id of the country. 

125 iso_2: The ISO 2 code of the country. 

126 iso_3: The ISO 3 code of the country. 

127 """ 

128 

129 __table_name__ = "countries" 

130 

131 id: int | None 

132 iso_2: str | None 

133 iso_3: str | None 

134 name: str | None 

135 created_at: datetime | None 

136 updated_at: datetime | None 

137 

138 def create_country(self) -> CountryEntity | None: 

139 """Create a Country value object from the row. 

140 

141 Returns: 

142 A country value object. 

143 """ 

144 if self.id is None: 

145 return None 

146 

147 return CountryEntity( 

148 id=CountryIdentifier(self.id), 

149 iso_2=self.iso_2, # type: ignore 

150 iso_3=self.iso_3, # type: ignore 

151 name=self.name, # type: ignore 

152 )