Coverage for bc/kwai-bc-portal/src/kwai_bc_portal/repositories/author_db_query.py: 100%

1"""Module that implements an AuthorQuery for a database.""" 

2 

3from dataclasses import dataclass 

4from typing import Self 

5 

6from kwai_core.db.database import Database 

7from kwai_core.db.database_query import DatabaseQuery 

8from kwai_core.db.table_row import JoinedTableRow 

9from kwai_core.domain.value_objects.unique_id import UniqueId 

10from sql_smith.functions import on 

11 

12from kwai_bc_portal.domain.author import AuthorEntity, AuthorIdentifier 

13from kwai_bc_portal.repositories._tables import AuthorRow, UserRow 

14from kwai_bc_portal.repositories.author_query import AuthorQuery 

15 

16 

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

18class AuthorQueryRow(JoinedTableRow): 

19 """A data transfer object for the author query.""" 

20 

21 author: AuthorRow 

22 user: UserRow 

23 

24 def create_entity(self) -> AuthorEntity: 

25 """Create an Author entity from a row.""" 

26 return self.author.create_entity( 

27 uuid=UniqueId.create_from_string(self.user.uuid) 

28 ) 

29 

30 

31class AuthorDbQuery(AuthorQuery, DatabaseQuery): 

32 """An author query using a database.""" 

33 

34 def __init__(self, database: Database): 

35 super().__init__(database) 

36 

37 def init(self): 

38 self._query.from_(AuthorRow.__table_name__).join( 

39 UserRow.__table_name__, 

40 on(UserRow.column("id"), AuthorRow.column("user_id")), 

41 ) 

42 

43 @property 

44 def columns(self): 

45 return AuthorQueryRow.get_aliases() 

46 

47 def filter_by_id(self, id_: AuthorIdentifier) -> Self: 

48 self._query.and_where(AuthorRow.field("user_id").eq(id_.value)) 

49 return self 

50 

51 def filter_by_uuid(self, uuid: UniqueId): 

52 self._query.and_where(UserRow.field("uuid").eq(uuid)) 

53 return self