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

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

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

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

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

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

12 UserRecoveryEntity, 

13 UserRecoveryIdentifier, 

14) 

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

16 

17 

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

19class UserRecoveryRow(TableRow): 

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

21 

22 __table_name__ = "user_recoveries" 

23 

24 id: int | None 

25 user_id: int 

26 uuid: str 

27 expired_at: datetime 

28 confirmed_at: datetime | None 

29 mailed_at: datetime | None 

30 remark: str | None 

31 created_at: datetime 

32 updated_at: datetime | None 

33 

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

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

36 return UserRecoveryEntity( 

37 id=UserRecoveryIdentifier(self.id), 

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

39 user=user, 

40 expiration=Timestamp.create_utc(self.expired_at), 

41 remark=self.remark, 

42 confirmation=Timestamp.create_utc(self.confirmed_at), 

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

44 traceable_time=TraceableTime( 

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

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

47 ), 

48 ) 

49 

50 @classmethod 

51 def persist(cls, user_recovery: UserRecoveryEntity) -> Self: 

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

53 return UserRecoveryRow( 

54 id=user_recovery.id.value, 

55 user_id=user_recovery.user.id.value, 

56 uuid=str(user_recovery.uuid), 

57 expired_at=user_recovery.expiration.timestamp, 

58 confirmed_at=user_recovery.confirmation.timestamp, 

59 mailed_at=user_recovery.mailed_at.timestamp, 

60 remark=user_recovery.remark, 

61 created_at=user_recovery.traceable_time.created_at.timestamp, 

62 updated_at=user_recovery.traceable_time.updated_at.timestamp, 

63 )