backend on fastapi

backend/routers/exercises.py

@@ -0,0 +1,99 @@
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from typing import List
+import database
+import models
+import schemas
+from datetime import datetime
+
+router = APIRouter(prefix="/exercises", tags=["exercises"])
+
+def get_current_timestamp():
+    return datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%fZ')
+
+@router.post("/", response_model=schemas.Exercise)
+def create_exercise(exercise: schemas.ExerciseCreate, db: Session = Depends(database.get_db)):
+    db_exercise = db.query(models.Exercise).filter(models.Exercise.sync_id == exercise.sync_id).first()
+    if db_exercise:
+        raise HTTPException(status_code=400, detail="Exercise with this sync_id already exists")
+
+    timestamp = get_current_timestamp()
+    db_exercise = models.Exercise(
+        **exercise.dict(),
+        created_at=timestamp,
+        updated_at=timestamp
+    )
+    db.add(db_exercise)
+    db.commit()
+    db.refresh(db_exercise)
+    return db_exercise
+
+@router.get("/", response_model=List[schemas.Exercise])
+def list_exercises(workout_type_id: int = None, db: Session = Depends(database.get_db)):
+    query = db.query(models.Exercise)
+    if workout_type_id:
+        query = query.filter(models.Exercise.workout_type_id == workout_type_id)
+    return query.order_by(models.Exercise.name).all()
+
+@router.get("/{exercise_id}", response_model=schemas.Exercise)
+def get_exercise(exercise_id: int, db: Session = Depends(database.get_db)):
+    db_exercise = db.query(models.Exercise).filter(models.Exercise.id == exercise_id).first()
+    if not db_exercise:
+        raise HTTPException(status_code=404, detail="Exercise not found")
+    return db_exercise
+
+@router.put("/{exercise_id}", response_model=schemas.Exercise)
+def update_exercise(exercise_id: int, exercise: schemas.ExerciseUpdate, db: Session = Depends(database.get_db)):
+    db_exercise = db.query(models.Exercise).filter(models.Exercise.id == exercise_id).first()
+    if not db_exercise:
+        raise HTTPException(status_code=404, detail="Exercise not found")
+
+    update_data = exercise.dict(exclude_unset=True)
+    for key, value in update_data.items():
+        setattr(db_exercise, key, value)
+
+    db_exercise.updated_at = get_current_timestamp()
+    db.commit()
+    db.refresh(db_exercise)
+    return db_exercise
+
+@router.delete("/{exercise_id}")
+def delete_exercise(exercise_id: int, db: Session = Depends(database.get_db)):
+    db_exercise = db.query(models.Exercise).filter(models.Exercise.id == exercise_id).first()
+    if not db_exercise:
+        raise HTTPException(status_code=404, detail="Exercise not found")
+    db.delete(db_exercise)
+    db.commit()
+    return {"status": "deleted"}
+
+@router.get("/history/{exercise_name}")
+def get_exercise_history(exercise_name: str, client_id: int = None, limit: int = 10, db: Session = Depends(database.get_db)):
+    """Получить историю упражнения для подсказок на фронте"""
+    query = db.query(
+        models.Workout.start_time,
+        models.Set.weight,
+        models.Set.reps,
+        models.Set.is_dropset
+    ).join(
+        models.WorkoutExercise, models.WorkoutExercise.workout_id == models.Workout.id
+    ).join(
+        models.Set, models.Set.workout_exercise_id == models.WorkoutExercise.id
+    ).filter(
+        models.WorkoutExercise.exercise_name == exercise_name,
+        models.Workout.end_time.isnot(None)
+    )
+
+    if client_id:
+        query = query.filter(models.Workout.client_id == client_id)
+
+    results = query.order_by(models.Workout.start_time.desc()).limit(limit).all()
+
+    return [
+        {
+            "date": r.start_time,
+            "weight": r.weight,
+            "reps": r.reps,
+            "is_dropset": bool(r.is_dropset)
+        }
+        for r in results
+    ]