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

28 statements  

« prev     ^ index     » next       coverage.py v7.8.0, created at 2024-01-01 00:00 +0000

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 editor: int 

35 created_at: datetime 

36 updated_at: datetime | None 

37 

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

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

40 return AuthorEntity( 

41 id=AuthorIdentifier(self.user_id), 

42 uuid=uuid, 

43 name=self.name, 

44 remark=self.remark, 

45 active=self.active == 1, 

46 editor=self.editor == 1, 

47 traceable_time=TraceableTime( 

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

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

50 ), 

51 ) 

52 

53 @classmethod 

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

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

56 return cls( 

57 user_id=author.id.value, 

58 name=author.name, 

59 remark=author.remark, 

60 active=1 if author.active else 0, 

61 editor=1 if author.editor else 0, 

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

63 updated_at=author.traceable_time.updated_at.timestamp, 

64 )