Coverage for src/kwai/modules/portal/get_news_items.py: 90%

1"""Implement the use case: get news items.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai.core.domain.use_case import UseCaseBrowseResult 

6from kwai.core.domain.value_objects.unique_id import UniqueId 

7from kwai.modules.portal.news.news_item_repository import NewsItemRepository 

8 

9 

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

11class GetNewsItemsCommand: 

12 """Input for use case: [GetNewsItems][kwai.modules.news.get_news_items.GetNewsItems]. 

13 

14 Attributes: 

15 offset: Offset to use. Default is None. 

16 limit: The max. number of elements to return. Default is None, which means all. 

17 enabled: When False, also news items that are not activated will be returned. 

18 """ 

19 

20 offset: int | None = None 

21 limit: int | None = None 

22 enabled: bool = True 

23 publish_year: int = 0 

24 publish_month: int = 0 

25 application: int | str | None = None 

26 promoted: bool = False 

27 author_uuid: str | None = None 

28 

29 

30class GetNewsItems: 

31 """Implementation of the use case. 

32 

33 Use this use case for getting news items. 

34 """ 

35 

36 def __init__(self, repo: NewsItemRepository): 

37 """Initialize the use case. 

38 

39 Args: 

40 repo: A repository for getting the news items. 

41 """ 

42 self._repo = repo 

43 

44 async def execute(self, command: GetNewsItemsCommand) -> UseCaseBrowseResult: 

45 """Execute the use case. 

46 

47 Args: 

48 command: The input for this use case. 

49 

50 Returns: 

51 A tuple with the number of entities and an iterator for news item entities. 

52 """ 

53 query = self._repo.create_query() 

54 

55 if command.enabled: 

56 query.filter_by_active() 

57 

58 if command.publish_year > 0: 

59 query.filter_by_publication_date( 

60 command.publish_year, command.publish_month 

61 ) 

62 

63 if command.promoted: 

64 query.filter_by_promoted() 

65 

66 if command.application is not None: 

67 query.filter_by_application(command.application) 

68 

69 if command.author_uuid is not None: 

70 query.filter_by_user(UniqueId.create_from_string(command.author_uuid)) 

71 

72 query.order_by_publication_date() 

73 

74 return UseCaseBrowseResult( 

75 count=await query.count(), 

76 iterator=self._repo.get_all( 

77 query=query, offset=command.offset, limit=command.limit 

78 ), 

79 )