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

1"""Module that implements a MemberQuery 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, criteria, func, group, literal, on 

11 

12from kwai_bc_club.domain.member import MemberEntity, MemberIdentifier 

13from kwai_bc_club.repositories._tables import ( 

14 CoachRow, 

15 ContactRow, 

16 CountryRow, 

17 MemberRow, 

18 PersonRow, 

19 UserRow, 

20) 

21from kwai_bc_club.repositories.member_query import MemberQuery 

22 

23 

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

25class MemberQueryRow(JoinedTableRow): 

26 """A data transfer object for the Member query.""" 

27 

28 member: MemberRow 

29 person: PersonRow 

30 nationality: CountryRow 

31 contact: ContactRow 

32 country: CountryRow 

33 user: UserRow 

34 

35 def create_entity(self) -> MemberEntity: 

36 """Create a Member entity from a row.""" 

37 return self.member.create_entity( 

38 person=self.person.create_entity( 

39 nationality=self.nationality.create_country(), 

40 contact=self.contact.create_entity( 

41 country=self.country.create_country() 

42 ), 

43 ), 

44 user=self.user.create_user(), 

45 ) 

46 

47 

48class MemberDbQuery(MemberQuery, DatabaseQuery): 

49 """A database query for members.""" 

50 

51 def __init__(self, database: Database): 

52 super().__init__(database) 

53 

54 @property 

55 def count_column(self): 

56 return MemberRow.column("id") 

57 

58 def init(self): 

59 self._query.from_(MemberRow.__table_name__).inner_join( 

60 PersonRow.__table_name__, 

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

62 ).inner_join( 

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

64 on( 

65 "nationality.id", 

66 PersonRow.column("nationality_id"), 

67 ), 

68 ).inner_join( 

69 ContactRow.__table_name__, 

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

71 ).inner_join( 

72 CountryRow.__table_name__, 

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

74 ).left_join( 

75 UserRow.__table_name__, 

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

77 ) 

78 

79 @property 

80 def columns(self): 

81 return MemberQueryRow.get_aliases() 

82 

83 def filter_by_id(self, id_: MemberIdentifier) -> Self: 

84 self._query.and_where(MemberRow.field("id").eq(id_.value)) 

85 return self 

86 

87 def filter_by_license(self, license: str) -> Self: 

88 self._query.and_where(MemberRow.field("license").eq(license)) 

89 return self 

90 

91 def filter_by_license_date( 

92 self, license_end_month: int, license_end_year: int 

93 ) -> Self: 

94 condition = criteria( 

95 "{} = {}", 

96 func("YEAR", MemberRow.column("license_end_date")), 

97 literal(license_end_year), 

98 ).and_( 

99 criteria( 

100 "{} = {}", 

101 func("MONTH", MemberRow.column("license_end_date")), 

102 literal(license_end_month), 

103 ), 

104 ) 

105 self._query.and_where(group(condition)) 

106 return self 

107 

108 def filter_by_active(self) -> Self: 

109 self._query.and_where(MemberRow.field("active").eq(1)) 

110 return self 

111 

112 def filter_by_uuid(self, uuid: UniqueId) -> Self: 

113 self._query.and_where(MemberRow.field("uuid").eq(str(uuid))) 

114 return self 

115 

116 def filter_like_name(self, name: str) -> Self: 

117 self._query.and_where( 

118 group( 

119 PersonRow.search("lastname") 

120 .contains(name) 

121 .or_(PersonRow.search("firstname").contains(name)) 

122 ) 

123 ) 

124 return self 

125 

126 def filter_is_coach(self) -> Self: 

127 coach_query = ( 

128 self._database.create_query_factory() 

129 .select("member_id") 

130 .from_(CoachRow.__table_name__) 

131 ) 

132 self._query.and_where(MemberRow.field("id").in_(coach_query)) 

133 return self 

134 

135 def filter_is_not_coach(self) -> Self: 

136 coach_query = ( 

137 self._database.create_query_factory() 

138 .select("member_id") 

139 .from_(CoachRow.__table_name__) 

140 ) 

141 self._query.and_where(MemberRow.field("id").not_in(coach_query)) 

142 return self