Coverage for apps/kwai-api/src/kwai_api/v1/training_schedules/endpoints.py: 83%
84 statements
« prev ^ index » next coverage.py v7.11.0, created at 2024-01-01 00:00 +0000
1"""Module for endpoints for training schedules."""
3from fastapi import APIRouter, Depends, HTTPException, status
4from kwai_bc_club.repositories.coach_repository import CoachNotFoundException
5from kwai_bc_identity.users.user import UserEntity
6from kwai_bc_training.coaches.coach_db_repository import CoachDbRepository
7from kwai_bc_training.create_training_schedule import (
8 CreateTrainingSchedule,
9 CreateTrainingScheduleCommand,
10)
11from kwai_bc_training.delete_training_schedule import (
12 DeleteTrainingSchedule,
13 DeleteTrainingScheduleCommand,
14)
15from kwai_bc_training.get_training_schedule import (
16 GetTrainingSchedule,
17 GetTrainingScheduleCommand,
18)
19from kwai_bc_training.get_training_schedules import (
20 GetTrainingSchedules,
21 GetTrainingSchedulesCommand,
22)
23from kwai_bc_training.get_trainings import GetTrainings, GetTrainingsCommand
24from kwai_bc_training.teams.team_db_repository import TeamDbRepository
25from kwai_bc_training.trainings.training_db_repository import TrainingDbRepository
26from kwai_bc_training.trainings.training_schedule_db_repository import (
27 TrainingScheduleDbRepository,
28)
29from kwai_bc_training.trainings.training_schedule_repository import (
30 TrainingScheduleNotFoundException,
31)
32from kwai_bc_training.update_training_schedule import (
33 UpdateTrainingSchedule,
34 UpdateTrainingScheduleCommand,
35)
36from kwai_core.db.uow import UnitOfWork
37from kwai_core.domain.value_objects.owner import Owner
38from kwai_core.json_api import PaginationModel
40from kwai_api.dependencies import create_database, get_current_user
41from kwai_api.v1.training_schedules.presenters import (
42 JsonApiTrainingSchedulePresenter,
43 JsonApiTrainingSchedulesPresenter,
44)
45from kwai_api.v1.training_schedules.schemas import (
46 TrainingScheduleDocument,
47 TrainingSchedulesDocument,
48)
49from kwai_api.v1.trainings.endpoints import TrainingsFilterModel
50from kwai_api.v1.trainings.presenters import JsonApiTrainingsDocumentPresenter
51from kwai_api.v1.trainings.schemas import TrainingsDocument
54router = APIRouter()
57@router.get("/training_schedules")
58async def get_training_schedules(
59 pagination: PaginationModel = Depends(PaginationModel),
60 db=Depends(create_database),
61) -> TrainingSchedulesDocument:
62 """Get all training schedules."""
63 command = GetTrainingSchedulesCommand(
64 offset=pagination.offset or 0, limit=pagination.limit
65 )
66 presenter = JsonApiTrainingSchedulesPresenter()
67 await GetTrainingSchedules(TrainingScheduleDbRepository(db), presenter).execute(
68 command
69 )
71 return presenter.get_document()
74@router.get(
75 "/training_schedules/{training_schedule_id}",
76 responses={
77 status.HTTP_404_NOT_FOUND: {"description": "Training schedule was not found."}
78 },
79)
80async def get_training_schedule(
81 training_schedule_id: int,
82 db=Depends(create_database),
83) -> TrainingScheduleDocument:
84 """Get training schedule with the given id."""
85 command = GetTrainingScheduleCommand(id=training_schedule_id)
86 presenter = JsonApiTrainingSchedulePresenter()
87 try:
88 await GetTrainingSchedule(TrainingScheduleDbRepository(db), presenter).execute(
89 command
90 )
91 except TrainingScheduleNotFoundException as ex:
92 raise HTTPException(
93 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
94 ) from ex
96 return presenter.get_document()
99@router.post(
100 "/training_schedules",
101 status_code=status.HTTP_201_CREATED,
102)
103async def create_training_schedule(
104 resource: TrainingScheduleDocument,
105 db=Depends(create_database),
106 user: UserEntity = Depends(get_current_user),
107) -> TrainingScheduleDocument:
108 """Create a new training schedule."""
109 if (
110 resource.data.relationships is not None
111 and resource.data.relationships.team is not None
112 and resource.data.relationships.team.data is not None
113 and resource.data.relationships.team.data.id is not None
114 ):
115 team_id = int(resource.data.relationships.team.data.id)
116 else:
117 team_id = None
119 coaches: list[str] = [
120 coach.id for coach in resource.data.relationships.coaches.data if coach.id
121 ]
123 command = CreateTrainingScheduleCommand(
124 name=resource.data.attributes.name,
125 description=resource.data.attributes.description,
126 weekday=resource.data.attributes.weekday,
127 start_time=resource.data.attributes.start_time,
128 end_time=resource.data.attributes.end_time,
129 timezone=resource.data.attributes.timezone,
130 active=resource.data.attributes.active,
131 location=resource.data.attributes.location or "",
132 remark=resource.data.attributes.remark or "",
133 team_id=team_id,
134 coaches=coaches,
135 )
136 presenter = JsonApiTrainingSchedulePresenter()
137 async with UnitOfWork(db):
138 try:
139 await CreateTrainingSchedule(
140 TrainingScheduleDbRepository(db),
141 TeamDbRepository(db),
142 CoachDbRepository(db),
143 Owner(id=user.id, uuid=user.uuid, name=user.name),
144 presenter,
145 ).execute(command)
146 except ValueError as ve:
147 raise HTTPException(
148 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ve)
149 ) from ve
151 return presenter.get_document()
154@router.patch(
155 "/training_schedules/{training_schedule_id}",
156 responses={
157 status.HTTP_404_NOT_FOUND: {"description": "Training schedule was not found."}
158 },
159)
160async def update_training_schedule(
161 training_schedule_id: int,
162 resource: TrainingScheduleDocument,
163 db=Depends(create_database),
164 user: UserEntity = Depends(get_current_user),
165) -> TrainingScheduleDocument:
166 """Update a training schedule."""
167 if (
168 resource.data.relationships is not None
169 and resource.data.relationships.team is not None
170 and resource.data.relationships.team.data is not None
171 and resource.data.relationships.team.data.id is not None
172 ):
173 team_id = int(resource.data.relationships.team.data.id)
174 else:
175 team_id = None
177 coaches = [coach.id for coach in resource.data.relationships.coaches.data]
179 command = UpdateTrainingScheduleCommand(
180 id=training_schedule_id,
181 name=resource.data.attributes.name,
182 description=resource.data.attributes.description,
183 weekday=resource.data.attributes.weekday,
184 start_time=resource.data.attributes.start_time,
185 end_time=resource.data.attributes.end_time,
186 timezone=resource.data.attributes.timezone,
187 active=resource.data.attributes.active,
188 location=resource.data.attributes.location or "",
189 remark=resource.data.attributes.remark or "",
190 team_id=team_id,
191 coaches=coaches,
192 )
193 presenter = JsonApiTrainingSchedulePresenter()
194 async with UnitOfWork(db):
195 try:
196 await UpdateTrainingSchedule(
197 TrainingScheduleDbRepository(db),
198 TeamDbRepository(db),
199 CoachDbRepository(db),
200 Owner(id=user.id, uuid=user.uuid, name=user.name),
201 presenter,
202 ).execute(command)
203 except TrainingScheduleNotFoundException as ex:
204 raise HTTPException(
205 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
206 ) from ex
207 except ValueError as ve:
208 raise HTTPException(
209 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(ve)
210 ) from ve
212 return presenter.get_document()
215@router.delete(
216 "/training_schedules/{training_schedule_id}",
217 responses={
218 status.HTTP_404_NOT_FOUND: {"description": "Training schedule was not found."}
219 },
220)
221async def delete_training_schedule(
222 training_schedule_id: int,
223 db=Depends(create_database),
224 user: UserEntity = Depends(get_current_user),
225) -> None:
226 """Delete a training schedule."""
227 command = DeleteTrainingScheduleCommand(
228 id=training_schedule_id, delete_trainings=False
229 )
230 await DeleteTrainingSchedule(
231 TrainingScheduleDbRepository(db), TrainingDbRepository(db)
232 ).execute(command)
235@router.get(
236 "/training_schedules/{training_schedule_id}/trainings",
237 responses={
238 status.HTTP_404_NOT_FOUND: {
239 "description": "Training schedule or coach was not found."
240 }
241 },
242)
243async def get_trainings(
244 training_schedule_id: int,
245 pagination: PaginationModel = Depends(PaginationModel),
246 trainings_filter: TrainingsFilterModel = Depends(TrainingsFilterModel),
247 db=Depends(create_database),
248) -> TrainingsDocument:
249 """Get trainings of the given training schedule."""
250 command = GetTrainingsCommand(
251 offset=pagination.offset or 0,
252 limit=pagination.limit,
253 year=trainings_filter.year,
254 month=trainings_filter.month,
255 start=trainings_filter.start,
256 end=trainings_filter.end,
257 active=trainings_filter.active,
258 coach_uuid=trainings_filter.coach,
259 schedule=training_schedule_id,
260 )
261 presenter = JsonApiTrainingsDocumentPresenter()
263 try:
264 await GetTrainings(
265 TrainingDbRepository(db),
266 CoachDbRepository(db),
267 TrainingScheduleDbRepository(db),
268 presenter,
269 ).execute(command)
270 except TrainingScheduleNotFoundException as ex:
271 raise HTTPException(
272 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
273 ) from ex
274 except CoachNotFoundException as ex:
275 raise HTTPException(
276 status_code=status.HTTP_404_NOT_FOUND, detail=str(ex)
277 ) from ex
279 return presenter.get_document()