295 lines
8.3 KiB
Python
295 lines
8.3 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session, joinedload
|
|
|
|
import auth
|
|
import database
|
|
import models
|
|
import schemas
|
|
|
|
|
|
router = APIRouter(
|
|
prefix="/workouts",
|
|
tags=["workouts"],
|
|
dependencies=[Depends(auth.get_current_user)],
|
|
)
|
|
|
|
|
|
def get_current_timestamp() -> str:
|
|
return datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%fZ")
|
|
|
|
|
|
def new_sync_id() -> str:
|
|
return uuid.uuid4().hex
|
|
|
|
|
|
def get_owned_workout(
|
|
db: Session,
|
|
workout_id: int,
|
|
user: models.User,
|
|
) -> models.Workout:
|
|
workout = (
|
|
db.query(models.Workout)
|
|
.filter(
|
|
models.Workout.id == workout_id,
|
|
models.Workout.owner_id == user.id,
|
|
)
|
|
.first()
|
|
)
|
|
if not workout:
|
|
raise HTTPException(status_code=404, detail="Workout not found")
|
|
return workout
|
|
|
|
|
|
def validate_client(
|
|
db: Session,
|
|
client_id: Optional[int],
|
|
user: models.User,
|
|
) -> Optional[int]:
|
|
if not client_id:
|
|
return None
|
|
|
|
client = (
|
|
db.query(models.Client)
|
|
.filter(
|
|
models.Client.id == client_id,
|
|
models.Client.owner_id == user.id,
|
|
)
|
|
.first()
|
|
)
|
|
if not client:
|
|
raise HTTPException(status_code=404, detail="Client not found")
|
|
return client.id
|
|
|
|
|
|
def validate_workout_type(
|
|
db: Session,
|
|
workout_type_id: Optional[int],
|
|
) -> Optional[int]:
|
|
if not workout_type_id:
|
|
return None
|
|
|
|
wt = db.get(models.WorkoutType, workout_type_id)
|
|
if not wt:
|
|
raise HTTPException(status_code=422, detail="Workout type not found")
|
|
return wt.id
|
|
|
|
|
|
def create_set_record(
|
|
db: Session,
|
|
workout_exercise_id: int,
|
|
set_data: schemas.SetCreate,
|
|
timestamp: str,
|
|
) -> models.Set:
|
|
is_dropset = 1 if set_data.is_dropset else 0
|
|
|
|
if is_dropset:
|
|
if set_data.weight is not None or set_data.reps is not None:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Dropset must not have top-level weight/reps",
|
|
)
|
|
if not set_data.dropset_items:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Dropset must have at least one dropset item",
|
|
)
|
|
else:
|
|
if set_data.weight is None or set_data.reps is None:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail="Regular set must have weight and reps",
|
|
)
|
|
|
|
db_set = models.Set(
|
|
sync_id=set_data.sync_id or new_sync_id(),
|
|
workout_exercise_id=workout_exercise_id,
|
|
is_dropset=is_dropset,
|
|
weight=None if is_dropset else float(set_data.weight),
|
|
reps=None if is_dropset else int(set_data.reps),
|
|
rating=set_data.rating or 3,
|
|
notes=set_data.notes or "",
|
|
sort_order=set_data.sort_order or 0,
|
|
timestamp=set_data.timestamp or timestamp,
|
|
created_at=timestamp,
|
|
)
|
|
|
|
db.add(db_set)
|
|
db.flush()
|
|
|
|
if is_dropset:
|
|
for idx, item in enumerate(set_data.dropset_items or []):
|
|
db.add(
|
|
models.DropsetItem(
|
|
set_id=db_set.id,
|
|
weight=float(item.weight),
|
|
reps=int(item.reps),
|
|
sort_order=item.sort_order if item.sort_order is not None else idx,
|
|
)
|
|
)
|
|
|
|
return db_set
|
|
|
|
|
|
@router.post("/", response_model=schemas.WorkoutOut)
|
|
def create_workout(
|
|
workout: schemas.WorkoutCreate,
|
|
current_user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(database.get_db),
|
|
):
|
|
timestamp = get_current_timestamp()
|
|
sync_id = workout.sync_id or new_sync_id()
|
|
|
|
existing = db.query(models.Workout).filter(models.Workout.sync_id == sync_id).first()
|
|
if existing:
|
|
if existing.owner_id != current_user.id:
|
|
raise HTTPException(status_code=409, detail="Workout sync_id belongs to another user")
|
|
return existing
|
|
|
|
client_id = validate_client(db, workout.client_id, current_user)
|
|
workout_type_id = validate_workout_type(db, workout.workout_type_id)
|
|
|
|
db_workout = models.Workout(
|
|
sync_id=sync_id,
|
|
owner_id=current_user.id,
|
|
client_id=client_id,
|
|
workout_type_id=workout_type_id,
|
|
start_time=workout.start_time,
|
|
end_time=workout.end_time,
|
|
notes=workout.notes or "",
|
|
synced=1,
|
|
created_at=timestamp,
|
|
updated_at=timestamp,
|
|
)
|
|
|
|
db.add(db_workout)
|
|
db.flush()
|
|
|
|
for we_data in workout.workout_exercises or []:
|
|
db_we = models.WorkoutExercise(
|
|
sync_id=we_data.sync_id or new_sync_id(),
|
|
workout_id=db_workout.id,
|
|
exercise_id=we_data.exercise_id,
|
|
exercise_name=we_data.exercise_name,
|
|
sort_order=we_data.sort_order or 0,
|
|
created_at=timestamp,
|
|
)
|
|
db.add(db_we)
|
|
db.flush()
|
|
|
|
for set_data in we_data.sets or []:
|
|
create_set_record(db, db_we.id, set_data, timestamp)
|
|
|
|
for cardio_data in workout.cardio_sessions or []:
|
|
db_cardio = models.Cardio(
|
|
sync_id=cardio_data.sync_id or new_sync_id(),
|
|
workout_id=db_workout.id,
|
|
cardio_type=cardio_data.cardio_type,
|
|
purpose=cardio_data.purpose,
|
|
duration=cardio_data.duration,
|
|
distance=cardio_data.distance,
|
|
speed=cardio_data.speed,
|
|
incline=cardio_data.incline,
|
|
resistance=cardio_data.resistance,
|
|
notes=cardio_data.notes or "",
|
|
sort_order=cardio_data.sort_order or 0,
|
|
created_at=timestamp,
|
|
)
|
|
db.add(db_cardio)
|
|
|
|
db.commit()
|
|
db.refresh(db_workout)
|
|
return db_workout
|
|
|
|
|
|
@router.get("/", response_model=List[schemas.WorkoutOut])
|
|
def list_workouts(
|
|
client_id: Optional[int] = None,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
current_user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(database.get_db),
|
|
):
|
|
query = db.query(models.Workout).options(
|
|
joinedload(models.Workout.workout_exercises)
|
|
.joinedload(models.WorkoutExercise.sets)
|
|
.joinedload(models.Set.dropset_items),
|
|
joinedload(models.Workout.cardio_sessions),
|
|
).filter(
|
|
models.Workout.owner_id == current_user.id
|
|
)
|
|
|
|
if client_id:
|
|
validate_client(db, client_id, current_user)
|
|
query = query.filter(models.Workout.client_id == client_id)
|
|
|
|
return query.order_by(models.Workout.start_time.desc()).offset(skip).limit(limit).all()
|
|
|
|
|
|
@router.get("/{workout_id}", response_model=schemas.WorkoutOut)
|
|
def get_workout(
|
|
workout_id: int,
|
|
current_user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(database.get_db),
|
|
):
|
|
db_workout = (
|
|
db.query(models.Workout)
|
|
.options(
|
|
joinedload(models.Workout.workout_exercises)
|
|
.joinedload(models.WorkoutExercise.sets)
|
|
.joinedload(models.Set.dropset_items),
|
|
joinedload(models.Workout.cardio_sessions),
|
|
)
|
|
.filter(
|
|
models.Workout.id == workout_id,
|
|
models.Workout.owner_id == current_user.id,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not db_workout:
|
|
raise HTTPException(status_code=404, detail="Workout not found")
|
|
|
|
return db_workout
|
|
|
|
|
|
@router.put("/{workout_id}", response_model=schemas.WorkoutOut)
|
|
def update_workout(
|
|
workout_id: int,
|
|
workout: schemas.WorkoutUpdate,
|
|
current_user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(database.get_db),
|
|
):
|
|
db_workout = get_owned_workout(db, workout_id, current_user)
|
|
|
|
update_data = workout.dict(exclude_unset=True)
|
|
|
|
if "workout_type_id" in update_data:
|
|
update_data["workout_type_id"] = validate_workout_type(
|
|
db,
|
|
update_data["workout_type_id"],
|
|
)
|
|
|
|
for key, value in update_data.items():
|
|
setattr(db_workout, key, value)
|
|
|
|
db_workout.updated_at = get_current_timestamp()
|
|
db.commit()
|
|
db.refresh(db_workout)
|
|
return db_workout
|
|
|
|
|
|
@router.delete("/{workout_id}")
|
|
def delete_workout(
|
|
workout_id: int,
|
|
current_user: models.User = Depends(auth.get_current_user),
|
|
db: Session = Depends(database.get_db),
|
|
):
|
|
db_workout = get_owned_workout(db, workout_id, current_user)
|
|
db.delete(db_workout)
|
|
db.commit()
|
|
return {"status": "deleted"}
|