Coverage for src/kwai/modules/teams/repositories/member_repository.py: 100%

1"""Module for defining the member repository interface.""" 

2 

3from abc import ABC, abstractmethod 

4from typing import AsyncGenerator, Self 

5 

6from kwai.core.domain.repository.query import Query 

7from kwai.core.domain.value_objects.date import Date 

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

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

10from kwai.modules.teams.domain.team_member import MemberEntity, MemberIdentifier 

11 

12 

13class MemberNotFoundException(Exception): 

14 """Raised when a member does not exist.""" 

15 

16 

17class MemberQuery(Query, ABC): 

18 """An interface for a member query.""" 

19 

20 @abstractmethod 

21 def filter_by_id(self, id_: MemberIdentifier) -> Self: 

22 """Find a team member by its id.""" 

23 raise NotImplementedError 

24 

25 @abstractmethod 

26 def filter_by_uuid(self, uuid: UniqueId) -> Self: 

27 """Find a team member by its uuid.""" 

28 raise NotImplementedError 

29 

30 @abstractmethod 

31 def filter_by_birthdate( 

32 self, start_date: Date, end_date: Date | None = None 

33 ) -> Self: 

34 """Find team members by their birthdate.""" 

35 raise NotImplementedError 

36 

37 @abstractmethod 

38 def filter_by_team(self, team_id: TeamIdentifier, in_team: bool = True) -> Self: 

39 """Find members that are (not) part of the team. 

40 

41 To get only the members that are not part of the team, set in_team to False. 

42 

43 Args: 

44 team_id: The id of the team 

45 in_team: Whether the member should be part of the team 

46 """ 

47 raise NotImplementedError 

48 

49 

50class MemberRepository(ABC): 

51 """An interface for a member repository.""" 

52 

53 @abstractmethod 

54 def create_query(self) -> MemberQuery: 

55 """Create a query for querying team members.""" 

56 raise NotImplementedError 

57 

58 @abstractmethod 

59 async def get(self, query: MemberQuery | None = None) -> MemberEntity: 

60 """Return the first returned element of the given query. 

61 

62 Args: 

63 query: The query to use for getting the first member. 

64 

65 Raises: 

66 MemberNotFoundException: If the member is not found. 

67 """ 

68 raise NotImplementedError 

69 

70 @abstractmethod 

71 def get_all( 

72 self, 

73 query: MemberQuery | None = None, 

74 limit: int | None = None, 

75 offset: int | None = None, 

76 ) -> AsyncGenerator[MemberEntity, None]: 

77 """Return all members of the given query. 

78 

79 Args: 

80 query: The query to use for getting the members. 

81 limit: The maximum number of members to return. 

82 offset: The offset to use for fetching members. 

83 

84 Yields: 

85 A team member entity. 

86 """ 

87 raise NotImplementedError