Coverage for apps/kwai-api/src/kwai_api/v1/training_schedules/presenters.py: 100%

1"""Module that defines presenters for the training_schedules endpoint.""" 

2 

3from typing import Self 

4 

5from kwai_bc_training.trainings.training_schedule import TrainingScheduleEntity 

6from kwai_core.domain.presenter import AsyncPresenter, IterableResult, Presenter 

7from kwai_core.json_api import ( 

8 JsonApiPresenter, 

9 Meta, 

10 Relationship, 

11 RelationshipList, 

12 ResourceMeta, 

13) 

14 

15from kwai_api.schemas.resources import CoachResourceIdentifier, TeamResourceIdentifier 

16from kwai_api.v1.training_schedules.schemas import ( 

17 CoachAttributes, 

18 CoachResource, 

19 TeamAttributes, 

20 TeamResource, 

21 TrainingScheduleAttributes, 

22 TrainingScheduleDocument, 

23 TrainingScheduleRelationships, 

24 TrainingScheduleResource, 

25 TrainingSchedulesDocument, 

26) 

27 

28 

29def _create_training_schedule_document( 

30 training_schedule: TrainingScheduleEntity, 

31) -> TrainingScheduleDocument: 

32 included: set[TeamResource | CoachResource] = ( 

33 set() 

34 if training_schedule.team is None 

35 else { 

36 TeamResource( 

37 id=str(training_schedule.team.id), 

38 attributes=TeamAttributes(name=training_schedule.team.name), 

39 ) 

40 } 

41 ) 

42 for coach in training_schedule.coaches: 

43 included.add( 

44 CoachResource( 

45 id=str(coach.id), attributes=CoachAttributes(name=str(coach.name)) 

46 ) 

47 ) 

48 

49 return TrainingScheduleDocument( 

50 data=TrainingScheduleResource( 

51 id=str(training_schedule.id), 

52 meta=ResourceMeta( 

53 created_at=str(training_schedule.traceable_time.created_at), 

54 updated_at=str(training_schedule.traceable_time.updated_at), 

55 ), 

56 attributes=TrainingScheduleAttributes( 

57 name=training_schedule.name, 

58 description=training_schedule.description, 

59 weekday=training_schedule.weekday.value, 

60 start_time=str(training_schedule.period.start), 

61 end_time=str(training_schedule.period.end), 

62 timezone=training_schedule.period.timezone, 

63 active=training_schedule.active, 

64 location=training_schedule.location, 

65 remark=training_schedule.remark or "", 

66 ), 

67 relationships=TrainingScheduleRelationships( 

68 team=Relationship[TeamResourceIdentifier]( 

69 data=None 

70 if training_schedule.team is None 

71 else TeamResourceIdentifier(id=str(training_schedule.team.id)) 

72 ), 

73 coaches=RelationshipList[CoachResourceIdentifier]( 

74 data=[ 

75 CoachResourceIdentifier(id=str(coach.id)) 

76 for coach in training_schedule.coaches 

77 ] 

78 ), 

79 ), 

80 ), 

81 included=included, 

82 ) 

83 

84 

85class JsonApiTrainingSchedulePresenter( 

86 JsonApiPresenter[TrainingScheduleDocument], Presenter[TrainingScheduleEntity] 

87): 

88 """A presenter for transforming a training schedule entity into a JSON:API document.""" 

89 

90 def present(self, training_schedule: TrainingScheduleEntity) -> Self: 

91 self._document = _create_training_schedule_document(training_schedule) 

92 return self 

93 

94 

95class JsonApiTrainingSchedulesPresenter( 

96 JsonApiPresenter[TrainingSchedulesDocument], 

97 AsyncPresenter[IterableResult[TrainingScheduleEntity]], 

98): 

99 """A presenter that transforms an iterator of training schedules into a JSON:API document.""" 

100 

101 async def present(self, result: IterableResult[TrainingScheduleEntity]) -> Self: 

102 self._document = TrainingSchedulesDocument( 

103 meta=Meta(count=result.count, offset=result.offset, limit=result.limit) 

104 ) 

105 async for training_schedule in result.iterator: 

106 training_schedule_document = _create_training_schedule_document( 

107 training_schedule 

108 ) 

109 if training_schedule_document.data is not None: 

110 self._document.data.append(training_schedule_document.data) 

111 self._document.included |= training_schedule_document.included 

112 return self