Coverage for apps/kwai-api/src/kwai_api/v1/trainings/teams/endpoints.py: 83%

1"""Module for defining the trainings/teams API.""" 

2 

3from typing import Annotated 

4 

5from fastapi import APIRouter, Depends, HTTPException, status 

6from kwai_bc_identity.users.user import UserEntity 

7from kwai_bc_training.add_team_to_training import ( 

8 AddTeamToTraining, 

9 AddTeamToTrainingCommand, 

10) 

11from kwai_bc_training.delete_team_from_training import ( 

12 DeleteTeamFromTraining, 

13 DeleteTeamFromTrainingCommand, 

14) 

15from kwai_bc_training.get_teams import GetTeams 

16from kwai_bc_training.get_training import GetTraining, GetTrainingCommand 

17from kwai_bc_training.teams.team_db_repository import TeamDbRepository 

18from kwai_bc_training.teams.team_repository import TeamNotFoundException 

19from kwai_bc_training.trainings.training_db_repository import TrainingDbRepository 

20from kwai_bc_training.trainings.training_repository import TrainingNotFoundException 

21from kwai_core.db.database import Database 

22from kwai_core.db.uow import UnitOfWork 

23from kwai_core.json_api import Meta, Relationship 

24 

25from kwai_api.dependencies import create_database, get_current_user 

26from kwai_api.schemas.resources import TeamResourceIdentifier 

27from kwai_api.v1.trainings.presenters import JsonApiTrainingPresenter 

28from kwai_api.v1.trainings.schemas import TrainingDocument 

29from kwai_api.v1.trainings.teams.presenters import JsonApiTrainingTeamsPresenter 

30from kwai_api.v1.trainings.teams.schemas import ( 

31 TeamAttributes, 

32 TeamResource, 

33 TeamsDocument, 

34) 

35 

36 

37router = APIRouter() 

38 

39 

40@router.get("/trainings/teams") 

41async def get_teams( 

42 database=Depends(create_database), 

43) -> TeamsDocument: 

44 """Get teams.""" 

45 count, team_iterator = await GetTeams(TeamDbRepository(database)).execute() 

46 

47 document = TeamsDocument(meta=Meta(count=count)) 

48 async for team in team_iterator: 

49 document.data.append( 

50 TeamResource( 

51 id=str(team.id), 

52 attributes=TeamAttributes(name=team.name), 

53 ) 

54 ) 

55 

56 return document 

57 

58 

59@router.get( 

60 "/trainings/{training_id}/teams", 

61 responses={status.HTTP_404_NOT_FOUND: {"description": "Training was not found."}}, 

62) 

63async def get_training_teams( 

64 training_id: int, database: Annotated[Database, Depends(create_database)] 

65) -> TeamsDocument: 

66 """Get the teams of the training.""" 

67 command = GetTrainingCommand(id=training_id) 

68 presenter = JsonApiTrainingTeamsPresenter() 

69 try: 

70 await GetTraining(TrainingDbRepository(database), presenter).execute(command) 

71 except TrainingNotFoundException as ex: 

72 raise HTTPException( 

73 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

74 ) from ex 

75 

76 return presenter.get_document() 

77 

78 

79@router.post( 

80 "/trainings/{training_id}/teams", 

81 responses={status.HTTP_404_NOT_FOUND: {"description": "Training was not found."}}, 

82 status_code=status.HTTP_201_CREATED, 

83) 

84async def add_team_to_training( 

85 resource: Relationship[TeamResourceIdentifier], 

86 training_id: int, 

87 database: Annotated[Database, Depends(create_database)], 

88 user: Annotated[UserEntity, Depends(get_current_user)], 

89) -> TrainingDocument: 

90 """Add a team to the training.""" 

91 if resource.data.id is None: 

92 raise HTTPException( 

93 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, 

94 detail="Id is missing for the teams relationship.", 

95 ) 

96 command = AddTeamToTrainingCommand( 

97 training_id=training_id, team_id=int(resource.data.id) 

98 ) 

99 presenter = JsonApiTrainingPresenter() 

100 try: 

101 async with UnitOfWork(database): 

102 await AddTeamToTraining( 

103 TrainingDbRepository(database), TeamDbRepository(database), presenter 

104 ).execute(command) 

105 except TrainingNotFoundException as ex: 

106 raise HTTPException( 

107 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

108 ) from ex 

109 except TeamNotFoundException as ex: 

110 raise HTTPException( 

111 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

112 ) from ex 

113 

114 return presenter.get_document() 

115 

116 

117@router.delete( 

118 "/trainings/{training_id}/teams/{team_id}", 

119 responses={status.HTTP_404_NOT_FOUND: {"description": "Training was not found."}}, 

120 status_code=status.HTTP_200_OK, 

121) 

122async def delete_team_from_training( 

123 training_id: int, 

124 team_id: int, 

125 database: Annotated[Database, Depends(create_database)], 

126 user: Annotated[UserEntity, Depends(get_current_user)], 

127) -> TrainingDocument: 

128 """Add a team to the training.""" 

129 command = DeleteTeamFromTrainingCommand(training_id=training_id, team_id=team_id) 

130 presenter = JsonApiTrainingPresenter() 

131 try: 

132 async with UnitOfWork(database): 

133 await DeleteTeamFromTraining( 

134 TrainingDbRepository(database), TeamDbRepository(database), presenter 

135 ).execute(command) 

136 except TrainingNotFoundException as ex: 

137 raise HTTPException( 

138 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

139 ) from ex 

140 except TeamNotFoundException as ex: 

141 raise HTTPException( 

142 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex) 

143 ) from ex 

144 

145 return presenter.get_document()