Coverage for bc/kwai-bc-club/src/kwai_bc_club/repositories/coach_db_query.py: 100%

1"""Module that implements the coach query interface for a database.""" 

2 

3from dataclasses import dataclass 

4from typing import Self 

5 

6from kwai_core.db.database import Database 

7from kwai_core.db.database_query import DatabaseQuery 

8from kwai_core.db.table_row import JoinedTableRow 

9from kwai_core.domain.value_objects.unique_id import UniqueId 

10from sql_smith.functions import alias, on 

11 

12from kwai_bc_club.domain.club_coach import ClubCoachEntity, ClubCoachIdentifier 

13from kwai_bc_club.domain.member import MemberIdentifier 

14from kwai_bc_club.repositories._tables import ( 

15 CoachRow, 

16 ContactRow, 

17 CountryRow, 

18 MemberRow, 

19 PersonRow, 

20 UserRow, 

21) 

22from kwai_bc_club.repositories.coach_query import CoachQuery 

23 

24 

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

26class CoachQueryRow(JoinedTableRow): 

27 """A datatransfer object for the coach query.""" 

28 

29 coach: CoachRow 

30 member: MemberRow 

31 person: PersonRow 

32 nationality: CountryRow 

33 contact: ContactRow 

34 country: CountryRow 

35 user: UserRow 

36 

37 def create_entity(self) -> ClubCoachEntity: 

38 """Create a club coach entity from a query row.""" 

39 return self.coach.create_entity( 

40 member=self.member.create_entity( 

41 person=self.person.create_entity( 

42 nationality=self.nationality.create_country(), 

43 contact=self.contact.create_entity( 

44 country=self.country.create_country() 

45 ), 

46 ), 

47 user=self.user.create_user(), 

48 ) 

49 ) 

50 

51 

52class CoachDBQuery(DatabaseQuery, CoachQuery): 

53 """A coach query implementation using a database.""" 

54 

55 def __init__(self, database: Database): 

56 super().__init__(database) 

57 

58 def init(self): 

59 ( 

60 self._query.from_(CoachRow.__table_name__) 

61 .inner_join( 

62 MemberRow.__table_name__, 

63 on(MemberRow.column("id"), CoachRow.column("member_id")), 

64 ) 

65 .inner_join( 

66 PersonRow.__table_name__, 

67 on(PersonRow.column("id"), MemberRow.column("person_id")), 

68 ) 

69 ).inner_join( 

70 alias(CountryRow.__table_name__, "nationality"), 

71 on( 

72 "nationality.id", 

73 PersonRow.column("nationality_id"), 

74 ), 

75 ).inner_join( 

76 ContactRow.__table_name__, 

77 on(ContactRow.column("id"), PersonRow.column("contact_id")), 

78 ).inner_join( 

79 CountryRow.__table_name__, 

80 on(CountryRow.column("id"), ContactRow.column("country_id")), 

81 ).left_join( 

82 UserRow.__table_name__, 

83 on(UserRow.column("id"), MemberRow.column("user_id")), 

84 ) 

85 

86 @property 

87 def count_column(self): 

88 return CoachRow.column("id") 

89 

90 @property 

91 def columns(self): 

92 return CoachQueryRow.get_aliases() 

93 

94 def filter_by_id(self, id_: ClubCoachIdentifier) -> Self: 

95 self._query.and_where(CoachRow.field("id").eq(id_.value)) 

96 return self 

97 

98 def filter_by_active(self) -> Self: 

99 self._query.and_where(CoachRow.field("active").eq(1)) 

100 return self 

101 

102 def filter_by_user_uuid(self, uuid: UniqueId) -> Self: 

103 self._query.and_where(UserRow.field("uuid").eq(str(uuid))) 

104 return self 

105 

106 def filter_by_member(self, member_id: MemberIdentifier) -> Self: 

107 self._query.and_where(MemberRow.field("id").eq(member_id)) 

108 return self