Coverage for bc/kwai-bc-teams/src/kwai_bc_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 

9 

10from kwai_bc_teams.domain.team import TeamIdentifier 

11from kwai_bc_teams.domain.team_member import MemberEntity, MemberIdentifier 

12 

13 

14class MemberNotFoundException(Exception): 

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

16 

17 

18class MemberQuery(Query, ABC): 

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

20 

21 @abstractmethod 

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

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

24 raise NotImplementedError 

25 

26 @abstractmethod 

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

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

29 raise NotImplementedError 

30 

31 @abstractmethod 

32 def filter_by_birthdate( 

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

34 ) -> Self: 

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

36 raise NotImplementedError 

37 

38 @abstractmethod 

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

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

41 

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

43 

44 Args: 

45 team_id: The id of the team 

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

47 """ 

48 raise NotImplementedError 

49 

50 

51class MemberRepository(ABC): 

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

53 

54 @abstractmethod 

55 def create_query(self) -> MemberQuery: 

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

57 raise NotImplementedError 

58 

59 @abstractmethod 

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

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

62 

63 Args: 

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

65 

66 Raises: 

67 MemberNotFoundException: If the member is not found. 

68 """ 

69 raise NotImplementedError 

70 

71 @abstractmethod 

72 def get_all( 

73 self, 

74 query: MemberQuery | None = None, 

75 limit: int | None = None, 

76 offset: int | None = None, 

77 ) -> AsyncGenerator[MemberEntity, None]: 

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

79 

80 Args: 

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

82 limit: The maximum number of members to return. 

83 offset: The offset to use for fetching members. 

84 

85 Yields: 

86 A team member entity. 

87 """ 

88 raise NotImplementedError