Coverage for kwai/modules/identity/user_recoveries/user_recovery_tables.py: 100%

1"""Module that defines the table for a user recovery.""" 

2from dataclasses import dataclass 

3from datetime import datetime 

4 

5from kwai.core.db.table import Table 

6from kwai.core.domain.value_objects.local_timestamp import LocalTimestamp 

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

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

9from kwai.modules.identity.user_recoveries.user_recovery import ( 

10 UserRecoveryEntity, 

11 UserRecoveryIdentifier, 

12) 

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

14 

15 

16@dataclass 

17class UserRecoveryRow: 

18 """Represent a row in the user recovery table.""" 

19 

20 id: int | None 

21 user_id: int 

22 uuid: str 

23 expired_at: datetime 

24 confirmed_at: datetime | None 

25 mailed_at: datetime | None 

26 remark: str | None 

27 created_at: datetime 

28 updated_at: datetime | None 

29 

30 def create_entity(self, user: UserEntity) -> UserRecoveryEntity: 

31 """Create a user recovery entity from the table row.""" 

32 return UserRecoveryEntity( 

33 id_=UserRecoveryIdentifier(self.id), 

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

35 user=user, 

36 expiration=LocalTimestamp(self.expired_at), 

37 remark=self.remark, 

38 confirmation=LocalTimestamp(self.confirmed_at), 

39 mailed_at=LocalTimestamp(self.mailed_at), 

40 traceable_time=TraceableTime( 

41 created_at=LocalTimestamp(timestamp=self.created_at), 

42 updated_at=LocalTimestamp(timestamp=self.updated_at), 

43 ), 

44 ) 

45 

46 @classmethod 

47 def persist(cls, user_recovery: UserRecoveryEntity) -> "UserRecoveryRow": 

48 """Map a user recovery entity to a table record.""" 

49 return UserRecoveryRow( 

50 id=user_recovery.id.value, 

51 user_id=user_recovery.user.id.value, 

52 uuid=str(user_recovery.uuid), 

53 expired_at=user_recovery.expiration.timestamp, 

54 confirmed_at=user_recovery.confirmed_at.timestamp, 

55 mailed_at=user_recovery.mailed_at.timestamp, 

56 remark=user_recovery.remark, 

57 created_at=user_recovery.traceable_time.created_at.timestamp, 

58 updated_at=user_recovery.traceable_time.updated_at.timestamp, 

59 ) 

60 

61 

62UserRecoveriesTable = Table("user_recoveries", UserRecoveryRow)