Coverage for bc/kwai-bc-training/src/kwai_bc_training/get_training_schedules.py: 100%

1"""Module that defines the use case "Get Training Schedules".""" 

2 

3from dataclasses import dataclass 

4 

5from kwai_core.domain.presenter import AsyncPresenter, IterableResult 

6 

7from kwai_bc_training.trainings.training_schedule import TrainingScheduleEntity 

8from kwai_bc_training.trainings.training_schedule_repository import ( 

9 TrainingScheduleRepository, 

10) 

11 

12 

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

14class GetTrainingSchedulesCommand: 

15 """Input for the use case.""" 

16 

17 limit: int | None = None 

18 offset: int | None = None 

19 

20 

21class GetTrainingSchedules: 

22 """Use case "Get Training Schedules".""" 

23 

24 def __init__(self, repo: TrainingScheduleRepository, presenter: AsyncPresenter): 

25 """Initialize the use case. 

26 

27 Args: 

28 repo: A repository for retrieving training schedules. 

29 presenter: A presenter for presenting the results. 

30 """ 

31 self._repo = repo 

32 self._presenter = presenter 

33 

34 async def execute(self, command: GetTrainingSchedulesCommand) -> None: 

35 """Execute the use case.""" 

36 query = self._repo.create_query() 

37 await self._presenter.present( 

38 IterableResult[TrainingScheduleEntity]( 

39 count=await query.count(), 

40 limit=command.limit or 0, 

41 offset=command.offset or 0, 

42 iterator=self._repo.get_all(query, command.limit, command.offset), 

43 ) 

44 )