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

1"""Module that defines an interface for a team repository.""" 

2 

3from abc import ABC, abstractmethod 

4from typing import AsyncGenerator, Self 

5 

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

7from kwai.modules.teams.domain.team import TeamEntity, TeamIdentifier 

8from kwai.modules.teams.domain.team_member import TeamMember 

9 

10 

11class TeamNotFoundException(Exception): 

12 """Raised when a team cannot be found.""" 

13 

14 

15class TeamQuery(Query, ABC): 

16 """An interface for a team query.""" 

17 

18 @abstractmethod 

19 def filter_by_id(self, id_: TeamIdentifier) -> Self: 

20 """Find a team by its id.""" 

21 raise NotImplementedError 

22 

23 

24class TeamRepository(ABC): 

25 """An interface for a team repository.""" 

26 

27 @abstractmethod 

28 def create_query(self) -> TeamQuery: 

29 """Create a team query.""" 

30 raise NotImplementedError 

31 

32 @abstractmethod 

33 async def get(self, query: TeamQuery | None = None) -> TeamEntity: 

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

35 

36 Args: 

37 query: The query to use for getting the first team. 

38 

39 Raises: 

40 TeamNotFoundException: If the team cannot be found. 

41 """ 

42 raise NotImplementedError 

43 

44 @abstractmethod 

45 def get_all( 

46 self, 

47 query: TeamQuery | None = None, 

48 limit: int | None = None, 

49 offset: int | None = None, 

50 ) -> AsyncGenerator[TeamEntity, None]: 

51 """Return all teams of the given query. 

52 

53 Args: 

54 query: The query to use for getting the teams. 

55 limit: The maximum number of teams to return. 

56 offset: The offset to use for fetching teams. 

57 

58 Yields: 

59 A team entity. 

60 

61 """ 

62 raise NotImplementedError 

63 

64 @abstractmethod 

65 async def create(self, team: TeamEntity) -> TeamEntity: 

66 """Save a new team.""" 

67 

68 @abstractmethod 

69 async def delete(self, team: TeamEntity) -> None: 

70 """Delete a team.""" 

71 

72 @abstractmethod 

73 async def update(self, team: TeamEntity) -> None: 

74 """Update a team.""" 

75 

76 @abstractmethod 

77 async def add_team_member(self, team: TeamEntity, member: TeamMember): 

78 """Add a member to a team."""