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 

38 def create_entity(self) -> UserEntity: 

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

40 return UserEntity( 

41 id=UserIdentifier(self.id), 

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

43 name=Name( 

44 first_name=self.first_name, 

45 last_name=self.last_name, 

46 ), 

47 remark=self.remark or "", 

48 email=EmailAddress(self.email), 

49 traceable_time=TraceableTime( 

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

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

52 ), 

53 ) 

54 

55 @classmethod 

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

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

58 return cls( 

59 id=user.id.value, 

60 email=str(user.email), 

61 first_name=user.name.first_name, 

62 last_name=user.name.last_name, 

63 remark=user.remark, 

64 uuid=str(user.uuid), 

65 created_at=user.traceable_time.created_at.timestamp, 

66 updated_at=user.traceable_time.updated_at.timestamp, 

67 person_id=None, 

68 ) 

69 

70 

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

72class UserAccountRow(TableRow): 

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

74 

75 __table_name__ = "users" 

76 

77 id: int | None 

78 email: str 

79 first_name: str 

80 last_name: str 

81 remark: str | None 

82 uuid: str 

83 created_at: datetime 

84 updated_at: datetime | None 

85 person_id: int | None 

86 last_login: datetime | None 

87 last_unsuccessful_login: datetime | None 

88 password: str 

89 revoked: int 

90 admin: int 

91 

92 def create_entity(self) -> UserAccountEntity: 

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

94 return UserAccountEntity( 

95 id=UserAccountIdentifier(self.id), 

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

97 last_login=Timestamp(self.last_login), 

98 last_unsuccessful_login=Timestamp(self.last_unsuccessful_login), 

99 revoked=self.revoked == 1, 

100 admin=self.admin == 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 ), 

115 ) 

116 

117 @classmethod 

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

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

120 return UserAccountRow( 

121 id=user_account.id.value, 

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

123 first_name=user_account.user.name.first_name, 

124 last_name=user_account.user.name.last_name, 

125 remark=user_account.user.remark, 

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

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

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

129 person_id=None, 

130 last_login=user_account.last_login.timestamp, 

131 last_unsuccessful_login=user_account.last_unsuccessful_login.timestamp, 

132 password=str(user_account.password), 

133 revoked=1 if user_account.revoked else 0, 

134 admin=1 if user_account.admin else 0, 

135 )