Coverage for bc/kwai-bc-teams/src/kwai_bc_teams/get_members.py: 100%

1"""Module for defining the use case 'Get Members'.""" 

2 

3from dataclasses import dataclass 

4 

5from kwai_core.domain.presenter import AsyncPresenter, IterableResult 

6 

7from kwai_bc_teams.domain.team import TeamIdentifier 

8from kwai_bc_teams.domain.team_member import MemberEntity 

9from kwai_bc_teams.repositories.member_repository import MemberRepository 

10 

11 

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

13class GetMembersCommand: 

14 """Input for the use case 'Get Members'. 

15 

16 When in_team is False and a team_id is set, only the members that are not 

17 part of that team will be returned. 

18 """ 

19 

20 team_id: int | None = None 

21 in_team: bool = True 

22 limit: int = 0 

23 offset: int = 0 

24 

25 

26class GetMembers: 

27 """Implements the use case 'Get Members'.""" 

28 

29 def __init__( 

30 self, 

31 member_repository: MemberRepository, 

32 presenter: AsyncPresenter[IterableResult[MemberEntity]], 

33 ): 

34 """Initialize the use case.""" 

35 self._member_repository = member_repository 

36 self._presenter = presenter 

37 

38 async def execute(self, command: GetMembersCommand): 

39 """Execute the use case.""" 

40 member_query = self._member_repository.create_query() 

41 if command.team_id is not None: 

42 member_query = member_query.filter_by_team( 

43 TeamIdentifier(command.team_id), command.in_team 

44 ) 

45 await self._presenter.present( 

46 IterableResult( 

47 count=await member_query.count(), 

48 limit=command.limit, 

49 offset=command.offset, 

50 iterator=self._member_repository.get_all( 

51 member_query, limit=command.limit, offset=command.offset 

52 ), 

53 ) 

54 )