Coverage for src/kwai/modules/identity/user_invitations/user_invitation_tables.py: 100%

1"""Module that defines all dataclasses for the tables containing invitations.""" 

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.timestamp import Timestamp 

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

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

13from kwai.modules.identity.user_invitations.user_invitation import ( 

14 UserInvitationEntity, 

15 UserInvitationIdentifier, 

16) 

17from kwai.modules.identity.users.user import UserEntity 

18 

19 

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

21class UserInvitationRow(TableRow): 

22 """Represent a table row in the invitations table. 

23 

24 Attributes: 

25 id(int): the id of the invitation 

26 email(str): the email that received this invitation 

27 first_name(str): the firstname of the invited 

28 last_name(str): the lastname of the invited 

29 uuid(str): a unique uuid for the invitation 

30 expired_at(datetime): the timestamp when the invitation expires 

31 remark(str|None): a remark about the invitation 

32 user_id(int): the user that created the invitation 

33 confirmed_at(datetime|None): the timestamp when the invitation was used 

34 revoked(bool): is the invitation revoked? 

35 created_at(datetime): the timestamp of creation 

36 updated_at(datetime|None): the timestamp of the last modification 

37 mailed_at(datetime|None): the timestamp of sending the email 

38 """ 

39 

40 __table_name__ = "user_invitations" 

41 

42 id: int 

43 email: str 

44 first_name: str 

45 last_name: str 

46 uuid: str 

47 expired_at: datetime 

48 remark: str | None 

49 user_id: int 

50 confirmed_at: datetime | None 

51 revoked: int 

52 created_at: datetime 

53 updated_at: datetime | None 

54 mailed_at: datetime | None 

55 

56 def create_entity(self, user: UserEntity) -> UserInvitationEntity: 

57 """Create a user invitation entity from the table row. 

58 

59 Args: 

60 user: The associated user entity 

61 

62 Returns: 

63 A user invitation entity. 

64 """ 

65 return UserInvitationEntity( 

66 id=UserInvitationIdentifier(self.id), 

67 email=EmailAddress(self.email), 

68 name=Name(last_name=self.last_name, first_name=self.first_name), 

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

70 expired_at=Timestamp.create_utc(self.expired_at), 

71 user=user, 

72 remark=self.remark or "", 

73 mailed_at=Timestamp.create_utc(self.mailed_at), 

74 confirmed_at=Timestamp.create_utc(self.confirmed_at), 

75 revoked=self.revoked == 1, 

76 traceable_time=TraceableTime( 

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

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

79 ), 

80 ) 

81 

82 @classmethod 

83 def persist(cls, invitation: UserInvitationEntity) -> Self: 

84 """Persist a user invitation entity into a table row. 

85 

86 Args: 

87 invitation: The user invitation entity to persist. 

88 

89 Returns: 

90 A dataclass containing the table row data. 

91 """ 

92 return UserInvitationRow( 

93 id=invitation.id.value, 

94 email=str(invitation.email), 

95 first_name=invitation.name.first_name, 

96 last_name=invitation.name.last_name, 

97 uuid=str(invitation.uuid), 

98 expired_at=invitation.expired_at.timestamp, 

99 mailed_at=invitation.mailed_at.timestamp, 

100 remark=invitation.remark, 

101 user_id=invitation.user.id.value, 

102 confirmed_at=invitation.confirmed_at.timestamp, 

103 revoked=1 if invitation.revoked else 0, 

104 created_at=invitation.traceable_time.created_at.timestamp, 

105 updated_at=invitation.traceable_time.updated_at.timestamp, 

106 )