Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/users/user_db_query.py: 100%

1"""Module that implements the UserQuery interface for a database.""" 

2 

3from kwai_core.db.database_query import DatabaseQuery 

4from kwai_core.domain.value_objects.email_address import EmailAddress 

5from kwai_core.domain.value_objects.unique_id import UniqueId 

6 

7from kwai_bc_identity.users.user import UserIdentifier 

8from kwai_bc_identity.users.user_query import UserQuery 

9from kwai_bc_identity.users.user_tables import UserRow 

10 

11 

12class UserDbQuery(UserQuery, DatabaseQuery): 

13 """A user query for a database.""" 

14 

15 def init(self): 

16 self._query.from_(UserRow.__table_name__) 

17 

18 @property 

19 def columns(self): 

20 return UserRow.get_aliases() 

21 

22 def filter_by_id(self, id_: UserIdentifier) -> UserQuery: 

23 """Add a filter for a user with the given id.""" 

24 self._query.and_where(UserRow.field("id").eq(id_.value)) 

25 return self 

26 

27 def filter_by_uuid(self, uuid: UniqueId) -> UserQuery: 

28 """Add a filter for a user with the given unique id.""" 

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

30 return self 

31 

32 def filter_by_email(self, email: EmailAddress) -> UserQuery: 

33 """Add a filter for a user with the given email address.""" 

34 self._query.and_where(UserRow.field("email").eq(str(email))) 

35 return self