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

11 

12from kwai_bc_identity.user_recoveries.user_recovery import ( 

13 UserRecoveryEntity, 

14 UserRecoveryIdentifier, 

15) 

16from kwai_bc_identity.users.user import UserEntity 

17 

18 

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

20class UserRecoveryRow(TableRow): 

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

22 

23 __table_name__ = "user_recoveries" 

24 

25 id: int | None 

26 user_id: int 

27 uuid: str 

28 expired_at: datetime 

29 confirmed_at: datetime | None 

30 mailed_at: datetime | None 

31 remark: str | None 

32 created_at: datetime 

33 updated_at: datetime | None 

34 

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

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

37 return UserRecoveryEntity( 

38 id=UserRecoveryIdentifier(self.id), 

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

40 user=user, 

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

42 remark=self.remark, 

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

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

45 traceable_time=TraceableTime( 

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

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

48 ), 

49 ) 

50 

51 @classmethod 

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

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

54 return UserRecoveryRow( 

55 id=user_recovery.id.value, 

56 user_id=user_recovery.user.id.value, 

57 uuid=str(user_recovery.uuid), 

58 expired_at=user_recovery.expiration.timestamp, 

59 confirmed_at=user_recovery.confirmation.timestamp, 

60 mailed_at=user_recovery.mailed_at.timestamp, 

61 remark=user_recovery.remark, 

62 created_at=user_recovery.traceable_time.created_at.timestamp, 

63 updated_at=user_recovery.traceable_time.updated_at.timestamp, 

64 )