Lars
1b7a0405e9
- Integrated exercise progression graphs into the backend and frontend, allowing users to visualize relationships between exercises. - Updated API endpoints for managing exercise progression graphs and edges, enhancing the exercise management capabilities. - Added a new tab in the ExercisesListPage for displaying progression graphs and included a panel in the ExerciseFormPage for editing. - Incremented application version to 0.8.6 and updated changelog to reflect new features and improvements.
347 lines
12 KiB
Python
347 lines
12 KiB
Python
"""
|
|
Progressionsgraph zwischen Übungen (Übung → Übung), Migration 032.
|
|
AuthZ analog training_plan_templates (_template_access / _has_planning_role).
|
|
"""
|
|
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from psycopg2 import IntegrityError
|
|
|
|
from auth import require_auth
|
|
from db import get_db, get_cursor, r2d
|
|
|
|
from routers.training_planning import _has_planning_role
|
|
|
|
router = APIRouter(prefix="/api", tags=["exercises"])
|
|
|
|
|
|
class ProgressionGraphCreate(BaseModel):
|
|
name: str = Field(..., min_length=1, max_length=200)
|
|
description: Optional[str] = None
|
|
visibility: str = Field(default="private", pattern="^(private|club|official)$")
|
|
club_id: Optional[int] = None
|
|
|
|
|
|
class ProgressionGraphUpdate(BaseModel):
|
|
name: Optional[str] = Field(None, min_length=1, max_length=200)
|
|
description: Optional[str] = None
|
|
visibility: Optional[str] = Field(None, pattern="^(private|club|official)$")
|
|
club_id: Optional[int] = None
|
|
|
|
|
|
class ProgressionEdgeCreate(BaseModel):
|
|
from_exercise_id: int = Field(..., gt=0)
|
|
to_exercise_id: int = Field(..., gt=0)
|
|
edge_type: str = Field(default="next_exercise", min_length=1, max_length=50)
|
|
notes: Optional[str] = Field(None, max_length=4000)
|
|
|
|
|
|
class ProgressionEdgeUpdate(BaseModel):
|
|
notes: Optional[str] = Field(None, max_length=4000)
|
|
|
|
|
|
_EDGE_SELECT = """
|
|
SELECT e.id, e.graph_id, e.from_exercise_id, e.to_exercise_id, e.edge_type, e.notes, e.created_at,
|
|
ef.title AS from_exercise_title, et.title AS to_exercise_title
|
|
FROM exercise_progression_edges e
|
|
JOIN exercises ef ON ef.id = e.from_exercise_id
|
|
JOIN exercises et ON et.id = e.to_exercise_id
|
|
"""
|
|
|
|
|
|
def _graph_access(cur, graph_id: int, profile_id: int, role: str) -> dict:
|
|
cur.execute(
|
|
"SELECT * FROM exercise_progression_graphs WHERE id = %s",
|
|
(graph_id,),
|
|
)
|
|
r = cur.fetchone()
|
|
if not r:
|
|
raise HTTPException(status_code=404, detail="Progressionsgraph nicht gefunden")
|
|
row = r2d(r)
|
|
if role in ("admin", "superadmin"):
|
|
return row
|
|
if row.get("created_by") != profile_id:
|
|
raise HTTPException(status_code=403, detail="Keine Berechtigung für diesen Progressionsgraph")
|
|
return row
|
|
|
|
|
|
def _assert_exercises_exist(cur, *exercise_ids: int) -> None:
|
|
unique_ids = list(dict.fromkeys(exercise_ids))
|
|
if not unique_ids:
|
|
return
|
|
cur.execute(
|
|
f"SELECT id FROM exercises WHERE id IN ({','.join(['%s'] * len(unique_ids))})",
|
|
tuple(unique_ids),
|
|
)
|
|
found = {r["id"] if isinstance(r, dict) else r[0] for r in cur.fetchall()}
|
|
missing = set(unique_ids) - found
|
|
if missing:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Unbekannte Übung(en): {sorted(missing)}",
|
|
)
|
|
|
|
|
|
@router.get("/exercise-progression-graphs")
|
|
def list_progression_graphs(session: dict = Depends(require_auth)):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if role in ("admin", "superadmin"):
|
|
cur.execute(
|
|
"""
|
|
SELECT g.*,
|
|
(SELECT COUNT(*) FROM exercise_progression_edges e WHERE e.graph_id = g.id) AS edges_count
|
|
FROM exercise_progression_graphs g
|
|
ORDER BY g.updated_at DESC NULLS LAST, g.name
|
|
"""
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT g.*,
|
|
(SELECT COUNT(*) FROM exercise_progression_edges e WHERE e.graph_id = g.id) AS edges_count
|
|
FROM exercise_progression_graphs g
|
|
WHERE g.created_by = %s
|
|
ORDER BY g.updated_at DESC NULLS LAST, g.name
|
|
""",
|
|
(profile_id,),
|
|
)
|
|
return [r2d(r) for r in cur.fetchall()]
|
|
|
|
|
|
@router.get("/exercise-progression-graphs/{graph_id}")
|
|
def get_progression_graph(
|
|
graph_id: int,
|
|
include_edges: bool = Query(default=False),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
row = _graph_access(cur, graph_id, profile_id, role)
|
|
if include_edges:
|
|
cur.execute(
|
|
_EDGE_SELECT + " WHERE e.graph_id = %s ORDER BY e.id",
|
|
(graph_id,),
|
|
)
|
|
row["edges"] = [r2d(r) for r in cur.fetchall()]
|
|
return row
|
|
|
|
|
|
@router.post("/exercise-progression-graphs", status_code=201)
|
|
def create_progression_graph(
|
|
body: ProgressionGraphCreate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
if not _has_planning_role(role):
|
|
raise HTTPException(status_code=403, detail="Keine Berechtigung zum Anlegen von Progressionsgraphen")
|
|
|
|
name = body.name.strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="name ist Pflicht")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO exercise_progression_graphs (name, description, visibility, club_id, created_by)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(name, body.description, body.visibility, body.club_id, profile_id),
|
|
)
|
|
gid = cur.fetchone()["id"]
|
|
conn.commit()
|
|
|
|
return get_progression_graph(gid, include_edges=False, session=session)
|
|
|
|
|
|
@router.put("/exercise-progression-graphs/{graph_id}")
|
|
def update_progression_graph(
|
|
graph_id: int,
|
|
body: ProgressionGraphUpdate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
data = body.model_dump(exclude_unset=True)
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="Keine Felder zum Aktualisieren")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
|
|
fields: List[str] = []
|
|
params: List[Any] = []
|
|
if "name" in data:
|
|
n = (data["name"] or "").strip()
|
|
if not n:
|
|
raise HTTPException(status_code=400, detail="name ist Pflicht")
|
|
fields.append("name = %s")
|
|
params.append(n)
|
|
if "description" in data:
|
|
fields.append("description = %s")
|
|
params.append(data["description"])
|
|
if "visibility" in data:
|
|
fields.append("visibility = %s")
|
|
params.append(data["visibility"])
|
|
if "club_id" in data:
|
|
fields.append("club_id = %s")
|
|
params.append(data["club_id"])
|
|
|
|
fields.append("updated_at = NOW()")
|
|
params.append(graph_id)
|
|
cur.execute(
|
|
f"UPDATE exercise_progression_graphs SET {', '.join(fields)} WHERE id = %s",
|
|
tuple(params),
|
|
)
|
|
conn.commit()
|
|
|
|
return get_progression_graph(graph_id, include_edges=False, session=session)
|
|
|
|
|
|
@router.delete("/exercise-progression-graphs/{graph_id}")
|
|
def delete_progression_graph(graph_id: int, session: dict = Depends(require_auth)):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
cur.execute("DELETE FROM exercise_progression_graphs WHERE id = %s", (graph_id,))
|
|
conn.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@router.get("/exercise-progression-graphs/{graph_id}/edges")
|
|
def list_progression_edges(
|
|
graph_id: int,
|
|
from_exercise_id: Optional[int] = Query(default=None),
|
|
to_exercise_id: Optional[int] = Query(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
q = _EDGE_SELECT + " WHERE e.graph_id = %s"
|
|
params: List[Any] = [graph_id]
|
|
if from_exercise_id is not None:
|
|
q += " AND e.from_exercise_id = %s"
|
|
params.append(from_exercise_id)
|
|
if to_exercise_id is not None:
|
|
q += " AND e.to_exercise_id = %s"
|
|
params.append(to_exercise_id)
|
|
q += " ORDER BY e.id"
|
|
cur.execute(q, tuple(params))
|
|
return [r2d(r) for r in cur.fetchall()]
|
|
|
|
|
|
@router.post("/exercise-progression-graphs/{graph_id}/edges", status_code=201)
|
|
def create_progression_edge(
|
|
graph_id: int,
|
|
body: ProgressionEdgeCreate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
if body.from_exercise_id == body.to_exercise_id:
|
|
raise HTTPException(status_code=400, detail="from_exercise_id und to_exercise_id müssen unterschiedlich sein")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
_assert_exercises_exist(cur, body.from_exercise_id, body.to_exercise_id)
|
|
et = (body.edge_type or "next_exercise").strip() or "next_exercise"
|
|
notes = (body.notes or "").strip() or None
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO exercise_progression_edges (
|
|
graph_id, from_exercise_id, to_exercise_id, edge_type, notes
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(graph_id, body.from_exercise_id, body.to_exercise_id, et, notes),
|
|
)
|
|
new_id = cur.fetchone()["id"]
|
|
cur.execute(_EDGE_SELECT + " WHERE e.id = %s", (new_id,))
|
|
row = r2d(cur.fetchone())
|
|
conn.commit()
|
|
except IntegrityError as e:
|
|
conn.rollback()
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="Kante existiert bereits (graph_id, von, nach, edge_type)",
|
|
) from e
|
|
|
|
return row
|
|
|
|
|
|
@router.put("/exercise-progression-graphs/{graph_id}/edges/{edge_id}")
|
|
def update_progression_edge(
|
|
graph_id: int,
|
|
edge_id: int,
|
|
body: ProgressionEdgeUpdate,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
data = body.model_dump(exclude_unset=True)
|
|
if not data:
|
|
raise HTTPException(status_code=400, detail="Keine Felder zum Aktualisieren")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
cur.execute(
|
|
"SELECT id FROM exercise_progression_edges WHERE id = %s AND graph_id = %s",
|
|
(edge_id, graph_id),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(status_code=404, detail="Kante nicht gefunden")
|
|
|
|
if "notes" in data:
|
|
n = data["notes"]
|
|
notes_val = (n or "").strip() or None if n is not None else None
|
|
cur.execute(
|
|
"UPDATE exercise_progression_edges SET notes = %s WHERE id = %s AND graph_id = %s",
|
|
(notes_val, edge_id, graph_id),
|
|
)
|
|
conn.commit()
|
|
cur.execute(_EDGE_SELECT + " WHERE e.id = %s AND e.graph_id = %s", (edge_id, graph_id))
|
|
return r2d(cur.fetchone())
|
|
|
|
|
|
@router.delete("/exercise-progression-graphs/{graph_id}/edges/{edge_id}")
|
|
def delete_progression_edge(
|
|
graph_id: int,
|
|
edge_id: int,
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
profile_id = session["profile_id"]
|
|
role = session.get("role")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
_graph_access(cur, graph_id, profile_id, role)
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM exercise_progression_edges
|
|
WHERE id = %s AND graph_id = %s
|
|
RETURNING id
|
|
""",
|
|
(edge_id, graph_id),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(status_code=404, detail="Kante nicht gefunden")
|
|
conn.commit()
|
|
return {"ok": True}
|