Coverage for src/kwai/modules/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 

6from kwai.modules.teams.domain.team import TeamIdentifier 

7from kwai.modules.teams.domain.team_member import MemberEntity 

8from kwai.modules.teams.repositories.member_repository import MemberRepository 

9 

10 

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

12class GetMembersCommand: 

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

14 

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

16 part of that team will be returned. 

17 """ 

18 

19 team_id: int | None = None 

20 in_team: bool = True 

21 limit: int | None = None 

22 offset: int | None = None 

23 

24 

25class GetMembers: 

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

27 

28 def __init__( 

29 self, 

30 member_repository: MemberRepository, 

31 presenter: AsyncPresenter[IterableResult[MemberEntity]], 

32 ): 

33 """Initialize the use case.""" 

34 self._member_repository = member_repository 

35 self._presenter = presenter 

36 

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

38 """Execute the use case.""" 

39 member_query = self._member_repository.create_query() 

40 if command.team_id is not None: 

41 member_query = member_query.filter_by_team( 

42 TeamIdentifier(command.team_id), command.in_team 

43 ) 

44 await self._presenter.present( 

45 IterableResult( 

46 count=await member_query.count(), 

47 limit=command.limit, 

48 offset=command.offset, 

49 iterator=self._member_repository.get_all(member_query), 

50 ) 

51 )