Coverage for kwai/modules/identity/users/user_db_repository.py: 89%

1"""Module for implementing a user repository with a database.""" 

2 

3from kwai.core.db.database import Database 

4from kwai.core.domain.value_objects.email_address import EmailAddress 

5from kwai.core.domain.value_objects.unique_id import UniqueId 

6from kwai.modules.identity.users.user import UserEntity, UserIdentifier 

7from kwai.modules.identity.users.user_db_query import UserDbQuery 

8from kwai.modules.identity.users.user_repository import ( 

9 UserNotFoundException, 

10 UserRepository, 

11) 

12from kwai.modules.identity.users.user_tables import UserRow, UsersTable 

13 

14 

15class UserDbRepository(UserRepository): 

16 """Database repository for the user entity.""" 

17 

18 def __init__(self, database: Database): 

19 self._database = database 

20 

21 async def update(self, user: UserEntity) -> None: 

22 await self._database.update( 

23 user.id.value, UsersTable.table_name, UserRow.persist(user) 

24 ) 

25 await self._database.commit() 

26 

27 def create_query(self) -> UserDbQuery: 

28 """Create a user database query.""" 

29 return UserDbQuery(self._database) 

30 

31 async def get_user_by_id(self, id_: UserIdentifier) -> UserEntity: 

32 """Get the user with the given id. 

33 

34 UserNotFoundException is raised when the user does not exist. 

35 """ 

36 query = self.create_query() 

37 query.filter_by_id(id_) 

38 

39 row = await query.fetch_one() 

40 if row: 

41 return UsersTable(row).create_entity() 

42 

43 raise UserNotFoundException() 

44 

45 async def get_user_by_uuid(self, uuid: UniqueId) -> UserEntity: 

46 """Get the user with the given uuid. 

47 

48 UserNotFoundException is raised when the user does not exist. 

49 """ 

50 query = self.create_query() 

51 query.filter_by_uuid(uuid) 

52 

53 row = await query.fetch_one() 

54 if row: 

55 return UsersTable(row).create_entity() 

56 

57 raise UserNotFoundException() 

58 

59 async def get_user_by_email(self, email: EmailAddress) -> UserEntity: 

60 """Get the user with the given email. 

61 

62 UserNotFoundException is raised when the user does not exist. 

63 """ 

64 query = self.create_query() 

65 query.filter_by_email(email) 

66 

67 row = await query.fetch_one() 

68 if row: 

69 return UsersTable(row).create_entity() 

70 

71 raise UserNotFoundException()