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

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

2 

3from dataclasses import dataclass 

4from datetime import datetime 

5from typing import Self 

6 

7from kwai.core.db.table_row import TableRow 

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

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

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

11from kwai.core.domain.value_objects.timestamp import Timestamp 

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

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

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

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

16 UserAccountEntity, 

17 UserAccountIdentifier, 

18) 

19 

20 

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

22class UserRow(TableRow): 

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

24 

25 __table_name__ = "users" 

26 

27 id: int | None 

28 email: str 

29 first_name: str 

30 last_name: str 

31 remark: str | None 

32 uuid: str 

33 created_at: datetime 

34 updated_at: datetime | None 

35 person_id: int | None 

36 

37 def create_entity(self) -> UserEntity: 

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

39 return UserEntity( 

40 id=UserIdentifier(self.id), 

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

42 name=Name( 

43 first_name=self.first_name, 

44 last_name=self.last_name, 

45 ), 

46 remark=self.remark, 

47 email=EmailAddress(self.email), 

48 traceable_time=TraceableTime( 

49 created_at=Timestamp.create_utc(timestamp=self.created_at), 

50 updated_at=Timestamp.create_utc(timestamp=self.updated_at), 

51 ), 

52 ) 

53 

54 @classmethod 

55 def persist(cls, user: UserEntity) -> Self: 

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

57 return cls( 

58 id=user.id.value, 

59 email=str(user.email), 

60 first_name=user.name.first_name, 

61 last_name=user.name.last_name, 

62 remark=user.remark, 

63 uuid=str(user.uuid), 

64 created_at=user.traceable_time.created_at.timestamp, 

65 updated_at=user.traceable_time.updated_at.timestamp, 

66 person_id=None, 

67 ) 

68 

69 

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

71class UserAccountRow(TableRow): 

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

73 

74 __table_name__ = "users" 

75 

76 id: int | None 

77 email: str 

78 first_name: str 

79 last_name: str 

80 remark: str | None 

81 uuid: str 

82 created_at: datetime 

83 updated_at: datetime | None 

84 person_id: int | None 

85 last_login: datetime | None 

86 last_unsuccessful_login: datetime | None 

87 password: str 

88 revoked: int 

89 admin: int 

90 

91 def create_entity(self) -> UserAccountEntity: 

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

93 return UserAccountEntity( 

94 id=UserAccountIdentifier(self.id), 

95 password=Password(self.password.encode()), 

96 last_login=Timestamp(self.last_login), 

97 last_unsuccessful_login=Timestamp(self.last_unsuccessful_login), 

98 revoked=self.revoked == 1, 

99 admin=self.admin == 1, 

100 user=UserEntity( 

101 id=UserIdentifier(self.id), 

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

103 name=Name( 

104 first_name=self.first_name, 

105 last_name=self.last_name, 

106 ), 

107 email=EmailAddress(self.email), 

108 traceable_time=TraceableTime( 

109 created_at=Timestamp.create_utc(self.created_at), 

110 updated_at=Timestamp.create_utc(self.updated_at), 

111 ), 

112 ), 

113 ) 

114 

115 @classmethod 

116 def persist(cls, user_account: UserAccountEntity) -> Self: 

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

118 return UserAccountRow( 

119 id=user_account.id.value, 

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

121 first_name=user_account.user.name.first_name, 

122 last_name=user_account.user.name.last_name, 

123 remark=None, 

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

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

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

127 person_id=None, 

128 last_login=user_account.last_login.timestamp, 

129 last_unsuccessful_login=user_account.last_unsuccessful_login.timestamp, 

130 password=str(user_account.password), 

131 revoked=1 if user_account.revoked else 0, 

132 admin=1 if user_account.admin else 0, 

133 )