Coverage for bc/kwai-bc-identity/src/kwai_bc_identity/users/user_tables.py: 96%

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 

14 

15from kwai_bc_identity.users.user import UserEntity, UserIdentifier 

16from kwai_bc_identity.users.user_account import ( 

17 UserAccountEntity, 

18 UserAccountIdentifier, 

19) 

20 

21 

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

23class UserRow(TableRow): 

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

25 

26 __table_name__ = "users" 

27 

28 id: int | None 

29 email: str 

30 first_name: str 

31 last_name: str 

32 remark: str | None 

33 uuid: str 

34 created_at: datetime 

35 updated_at: datetime | None 

36 person_id: int | None 

37 admin: int 

38 

39 def create_entity(self) -> UserEntity: 

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

41 return UserEntity( 

42 id=UserIdentifier(self.id), 

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

44 name=Name( 

45 first_name=self.first_name, 

46 last_name=self.last_name, 

47 ), 

48 remark=self.remark or "", 

49 email=EmailAddress(self.email), 

50 traceable_time=TraceableTime( 

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

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

53 ), 

54 ) 

55 

56 @classmethod 

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

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

59 return cls( 

60 id=user.id.value, 

61 email=str(user.email), 

62 first_name=user.name.first_name, 

63 last_name=user.name.last_name, 

64 remark=user.remark, 

65 uuid=str(user.uuid), 

66 created_at=user.traceable_time.created_at.timestamp, 

67 updated_at=user.traceable_time.updated_at.timestamp, 

68 person_id=None, 

69 ) 

70 

71 

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

73class UserAccountRow(TableRow): 

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

75 

76 __table_name__ = "users" 

77 

78 id: int | None 

79 email: str 

80 first_name: str 

81 last_name: str 

82 remark: str | None 

83 uuid: str 

84 created_at: datetime 

85 updated_at: datetime | None 

86 person_id: int | None 

87 last_login: datetime | None 

88 last_unsuccessful_login: datetime | None 

89 password: str 

90 revoked: int 

91 admin: int 

92 

93 def create_entity(self) -> UserAccountEntity: 

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

95 return UserAccountEntity( 

96 id=UserAccountIdentifier(self.id), 

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

98 last_login=Timestamp(self.last_login), 

99 last_unsuccessful_login=Timestamp(self.last_unsuccessful_login), 

100 revoked=self.revoked == 1, 

101 user=UserEntity( 

102 id=UserIdentifier(self.id), 

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

104 name=Name( 

105 first_name=self.first_name, 

106 last_name=self.last_name, 

107 ), 

108 email=EmailAddress(self.email), 

109 traceable_time=TraceableTime( 

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

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

112 ), 

113 remark=self.remark or "", 

114 admin=self.admin == 1, 

115 ), 

116 ) 

117 

118 @classmethod 

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

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

121 return UserAccountRow( 

122 id=user_account.id.value, 

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

124 first_name=user_account.user.name.first_name, 

125 last_name=user_account.user.name.last_name, 

126 remark=user_account.user.remark, 

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

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

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

130 person_id=None, 

131 last_login=user_account.last_login.timestamp, 

132 last_unsuccessful_login=user_account.last_unsuccessful_login.timestamp, 

133 password=str(user_account.password), 

134 revoked=1 if user_account.revoked else 0, 

135 admin=1 if user_account.user.admin else 0, 

136 )