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

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

2 

3from abc import ABC, abstractmethod 

4from typing import AsyncIterator 

5 

6from kwai.modules.training.teams.team import TeamEntity, TeamIdentifier 

7from kwai.modules.training.teams.team_query import TeamQuery 

8 

9 

10class TeamNotFoundException(Exception): 

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

12 

13 

14class TeamRepository(ABC): 

15 """Interface for a team repository.""" 

16 

17 @abstractmethod 

18 def create_query(self) -> TeamQuery: 

19 """Create a query for querying teams.""" 

20 raise NotImplementedError 

21 

22 @abstractmethod 

23 async def get_all(self) -> AsyncIterator[TeamEntity]: 

24 """Get all teams.""" 

25 

26 @abstractmethod 

27 async def get_by_id(self, id: TeamIdentifier) -> TeamEntity: 

28 """Get the team with the given id. 

29 

30 Args: 

31 id: An id of a team. 

32 """ 

33 raise NotImplementedError 

34 

35 @abstractmethod 

36 async def get_by_ids(self, *ids: TeamIdentifier) -> AsyncIterator[TeamEntity]: 

37 """Get all teams for the given ids. 

38 

39 Args: 

40 ids: A variable number of team ids. 

41 """ 

42 raise NotImplementedError