Coverage for bc/kwai-bc-identity/src/kwai_bc_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 

13 

14from kwai_bc_identity.user_invitations.user_invitation import ( 

15 UserInvitationEntity, 

16 UserInvitationIdentifier, 

17) 

18from kwai_bc_identity.users.user import UserEntity 

19 

20 

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

22class UserInvitationRow(TableRow): 

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

24 

25 Attributes: 

26 id(int): the id of the invitation 

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

28 first_name(str): the firstname of the invited 

29 last_name(str): the lastname of the invited 

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

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

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

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

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

35 revoked(bool): is the invitation revoked? 

36 created_at(datetime): the timestamp of creation 

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

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

39 """ 

40 

41 __table_name__ = "user_invitations" 

42 

43 id: int 

44 email: str 

45 first_name: str 

46 last_name: str 

47 uuid: str 

48 expired_at: datetime 

49 remark: str | None 

50 user_id: int 

51 confirmed_at: datetime | None 

52 revoked: int 

53 created_at: datetime 

54 updated_at: datetime | None 

55 mailed_at: datetime | None 

56 

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

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

59 

60 Args: 

61 user: The associated user entity 

62 

63 Returns: 

64 A user invitation entity. 

65 """ 

66 return UserInvitationEntity( 

67 id=UserInvitationIdentifier(self.id), 

68 email=EmailAddress(self.email), 

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

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

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

72 user=user, 

73 remark=self.remark or "", 

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

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

76 revoked=self.revoked == 1, 

77 traceable_time=TraceableTime( 

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

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

80 ), 

81 ) 

82 

83 @classmethod 

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

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

86 

87 Args: 

88 invitation: The user invitation entity to persist. 

89 

90 Returns: 

91 A dataclass containing the table row data. 

92 """ 

93 return UserInvitationRow( 

94 id=invitation.id.value, 

95 email=str(invitation.email), 

96 first_name=invitation.name.first_name, 

97 last_name=invitation.name.last_name, 

98 uuid=str(invitation.uuid), 

99 expired_at=invitation.expired_at.timestamp, 

100 mailed_at=invitation.mailed_at.timestamp, 

101 remark=invitation.remark, 

102 user_id=invitation.user.id.value, 

103 confirmed_at=invitation.confirmed_at.timestamp, 

104 revoked=1 if invitation.revoked else 0, 

105 created_at=invitation.traceable_time.created_at.timestamp, 

106 updated_at=invitation.traceable_time.updated_at.timestamp, 

107 )