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

11 

12from kwai_bc_portal.domain.author import AuthorEntity, AuthorIdentifier 

13 

14 

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

16class UserRow(TableRow): 

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

18 

19 __table_name__ = "users" 

20 

21 id: int 

22 uuid: str 

23 

24 

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

26class AuthorRow(TableRow): 

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

28 

29 __table_name__ = "authors" 

30 

31 user_id: int 

32 name: str 

33 remark: str 

34 active: int 

35 editor: int 

36 created_at: datetime 

37 updated_at: datetime | None 

38 

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

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

41 return AuthorEntity( 

42 id=AuthorIdentifier(self.user_id), 

43 uuid=uuid, 

44 name=self.name, 

45 remark=self.remark, 

46 active=self.active == 1, 

47 editor=self.editor == 1, 

48 traceable_time=TraceableTime( 

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

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

51 ), 

52 ) 

53 

54 @classmethod 

55 def persist(cls, author: AuthorEntity) -> Self: 

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

57 return cls( 

58 user_id=author.id.value, 

59 name=author.name, 

60 remark=author.remark, 

61 active=1 if author.active else 0, 

62 editor=1 if author.editor else 0, 

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

64 updated_at=author.traceable_time.updated_at.timestamp, 

65 )