Coverage for src/kwai/modules/portal/repositories/_tables.py: 100%

1"""Module for defining data transfer objects between domain and database.""" 

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.portal.domain.author import AuthorEntity, AuthorIdentifier 

12 

13 

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

15class UserRow(TableRow): 

16 """Represent a row in the users table.""" 

17 

18 __table_name__ = "users" 

19 

20 id: int 

21 uuid: str 

22 

23 

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

25class AuthorRow(TableRow): 

26 """Represent a row in the authors table.""" 

27 

28 __table_name__ = "authors" 

29 

30 user_id: int 

31 name: str 

32 remark: str 

33 active: int 

34 created_at: datetime 

35 updated_at: datetime | None 

36 

37 def create_entity(self, uuid: UniqueId) -> AuthorEntity: 

38 """Create an author entity from a table row.""" 

39 return AuthorEntity( 

40 id=AuthorIdentifier(self.user_id), 

41 uuid=uuid, 

42 name=self.name, 

43 remark=self.remark, 

44 active=self.active == 1, 

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, author: AuthorEntity) -> Self: 

53 """Transform an author entity into a table row.""" 

54 return cls( 

55 user_id=author.id.value, 

56 name=author.name, 

57 remark=author.remark, 

58 active=1 if author.active else 0, 

59 created_at=author.traceable_time.created_at.timestamp, # type: ignore 

60 updated_at=author.traceable_time.updated_at.timestamp, 

61 )