Coverage for kwai/modules/identity/users/user_tables.py: 98%

1"""Modules that defines all table classes for a user.""" 

2from dataclasses import dataclass 

3from datetime import datetime 

4 

5from kwai.core.db.table import Table 

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

7from kwai.core.domain.value_objects.local_timestamp import LocalTimestamp 

8from kwai.core.domain.value_objects.name import Name 

9from kwai.core.domain.value_objects.password import Password 

10from kwai.core.domain.value_objects.traceable_time import TraceableTime 

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

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

13from kwai.modules.identity.users.user_account import ( 

14 UserAccountEntity, 

15 UserAccountIdentifier, 

16) 

17 

18 

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

20class UserRow: 

21 """Represent a row in the users table.""" 

22 

23 id: int | None 

24 email: str 

25 first_name: str 

26 last_name: str 

27 remark: str | None 

28 uuid: str 

29 created_at: datetime 

30 updated_at: datetime | None 

31 person_id: int | None 

32 

33 def create_entity(self) -> UserEntity: 

34 """Create a user entity from a table row.""" 

35 return UserEntity( 

36 id_=UserIdentifier(self.id), 

37 uuid=UniqueId.create_from_string(self.uuid), 

38 name=Name( 

39 first_name=self.first_name, 

40 last_name=self.last_name, 

41 ), 

42 remark=self.remark, 

43 email=EmailAddress(self.email), 

44 traceable_time=TraceableTime( 

45 created_at=LocalTimestamp(timestamp=self.created_at), 

46 updated_at=LocalTimestamp(timestamp=self.updated_at), 

47 ), 

48 ) 

49 

50 @classmethod 

51 def persist(cls, user: UserEntity) -> "UserRow": 

52 """Transform a user entity into a table record.""" 

53 return UserRow( 

54 id=user.id.value, 

55 email=str(user.email), 

56 first_name=user.name.first_name, 

57 last_name=user.name.last_name, 

58 remark=user.remark, 

59 uuid=str(user.uuid), 

60 created_at=user.traceable_time.created_at.timestamp, 

61 updated_at=user.traceable_time.updated_at.timestamp, 

62 person_id=None, 

63 ) 

64 

65 

66UsersTable = Table("users", UserRow) 

67 

68 

69@dataclass(kw_only=True, frozen=True) 

70class UserAccountRow: 

71 """Represent a row in the user table with user account information.""" 

72 

73 id: int | None 

74 email: str 

75 first_name: str 

76 last_name: str 

77 remark: str | None 

78 uuid: str 

79 created_at: datetime 

80 updated_at: datetime | None 

81 person_id: int | None 

82 last_login: datetime | None 

83 last_unsuccessful_login: datetime | None 

84 password: str 

85 revoked: int 

86 admin: int 

87 

88 def create_entity(self) -> UserAccountEntity: 

89 """Create a user account entity from the table row.""" 

90 return UserAccountEntity( 

91 id_=UserAccountIdentifier(self.id), 

92 password=Password(hashed_password=self.password), 

93 last_login=LocalTimestamp(self.last_login), 

94 last_unsuccessful_login=LocalTimestamp(self.last_unsuccessful_login), 

95 revoked=self.revoked == 1, 

96 admin=self.admin == 1, 

97 user=UserEntity( 

98 id_=UserIdentifier(self.id), 

99 uuid=UniqueId.create_from_string(self.uuid), 

100 name=Name( 

101 first_name=self.first_name, 

102 last_name=self.last_name, 

103 ), 

104 email=EmailAddress(self.email), 

105 traceable_time=TraceableTime( 

106 created_at=LocalTimestamp(self.created_at), 

107 updated_at=LocalTimestamp(self.updated_at), 

108 ), 

109 ), 

110 ) 

111 

112 @classmethod 

113 def persist(cls, user_account: UserAccountEntity) -> "UserAccountRow": 

114 """Transform a user account entity into a table record.""" 

115 return UserAccountRow( 

116 id=user_account.id.value, 

117 email=str(user_account.user.email), 

118 first_name=user_account.user.name.first_name, 

119 last_name=user_account.user.name.last_name, 

120 remark=None, 

121 uuid=str(user_account.user.uuid), 

122 created_at=user_account.user.traceable_time.created_at.timestamp, 

123 updated_at=user_account.user.traceable_time.updated_at.timestamp, 

124 person_id=None, 

125 last_login=user_account.last_login.timestamp, 

126 last_unsuccessful_login=user_account.last_unsuccessful_login.timestamp, 

127 password=str(user_account.password), 

128 revoked=1 if user_account.revoked else 0, 

129 admin=1 if user_account.admin else 0, 

130 ) 

131 

132 

133UserAccountsTable = Table("users", UserAccountRow)