Lars
0275f76432
Some checks failed
Deploy Development / deploy (push) Successful in 40s
Test Suite / pytest-backend (push) Successful in 41s
Test Suite / lint-backend (push) Successful in 1s
Test Suite / build-frontend (push) Successful in 11s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Has been cancelled
- Introduced new functions for managing edit, delete, and governance transition permissions for library content, aligning with role-based access control (RBAC) principles. - Updated existing routers to utilize these new functions, ensuring consistent permission checks across training frameworks, modules, and progression graphs. - Enhanced visibility and governance handling for training plan templates and library content, improving overall content management and user experience. - Incremented app version to 0.8.142 and updated changelog to reflect these changes.
382 lines
13 KiB
Python
382 lines
13 KiB
Python
"""
|
||
Trainingsmodule — wiederverwendbare Planungsbausteine (Bibliothek).
|
||
|
||
Governance wie Trainings‑Mikrovorlagen (`training_plan_templates`):
|
||
Liste/Detail über `library_content_visibility_sql`; Bearbeiten/Löschen wie Übungen (s. club_tenancy).
|
||
|
||
Siehe `.claude/docs/working/TRAINING_MODULES_IMPLEMENTATION_PLAN.md`.
|
||
"""
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
|
||
from db import get_db, get_cursor, r2d
|
||
from tenant_context import TenantContext, get_tenant_context, library_content_visibility_sql
|
||
from club_tenancy import (
|
||
assert_library_content_deletable,
|
||
assert_library_content_editable,
|
||
assert_library_content_governance_transition,
|
||
assert_valid_governance_visibility,
|
||
is_platform_admin,
|
||
library_content_visible_to_profile,
|
||
)
|
||
|
||
router = APIRouter(prefix="/api", tags=["training_modules"])
|
||
|
||
|
||
def _has_planning_role(role: Optional[str]) -> bool:
|
||
return role in ("admin", "superadmin", "trainer", "user")
|
||
|
||
|
||
def _fetch_training_module_row(cur, mid: int) -> Dict[str, Any]:
|
||
cur.execute("SELECT * FROM training_modules WHERE id = %s", (mid,))
|
||
r = cur.fetchone()
|
||
if not r:
|
||
raise HTTPException(status_code=404, detail="Trainingsmodul nicht gefunden")
|
||
return r2d(r)
|
||
|
||
|
||
def _module_assert_readable(cur, row: Dict[str, Any], profile_id: int, role: Optional[str]) -> None:
|
||
if is_platform_admin(role):
|
||
return
|
||
if not library_content_visible_to_profile(
|
||
cur,
|
||
profile_id,
|
||
row.get("visibility") or "club",
|
||
row.get("club_id"),
|
||
row.get("created_by"),
|
||
role,
|
||
):
|
||
raise HTTPException(status_code=403, detail="Keine Berechtigung für dieses Modul")
|
||
|
||
|
||
def _module_access(cur, mid: int, profile_id: int, role: str) -> Dict[str, Any]:
|
||
row = _fetch_training_module_row(cur, mid)
|
||
_module_assert_readable(cur, row, profile_id, role)
|
||
return row
|
||
|
||
|
||
def _validate_variant_for_exercise(cur, exercise_id: Optional[int], variant_id: Optional[int]):
|
||
if not variant_id:
|
||
return
|
||
if not exercise_id:
|
||
raise HTTPException(
|
||
status_code=400, detail="exercise_variant_id nur zusammen mit exercise_id erlaubt"
|
||
)
|
||
cur.execute(
|
||
"SELECT 1 FROM exercise_variants WHERE id = %s AND exercise_id = %s",
|
||
(variant_id, exercise_id),
|
||
)
|
||
if not cur.fetchone():
|
||
raise HTTPException(status_code=400, detail="Variante passt nicht zur gewählten Übung")
|
||
|
||
|
||
def _optional_positive_int(val, field_name: str) -> Optional[int]:
|
||
if val is None or val == "":
|
||
return None
|
||
try:
|
||
i = int(val)
|
||
except (TypeError, ValueError):
|
||
raise HTTPException(status_code=400, detail=f"Ungültige {field_name}")
|
||
if i < 1:
|
||
raise HTTPException(status_code=400, detail=f"Ungültige {field_name}")
|
||
return i
|
||
|
||
|
||
def _replace_module_items(cur, module_id: int, items_in: Optional[List[Any]]) -> None:
|
||
cur.execute("DELETE FROM training_module_items WHERE module_id = %s", (module_id,))
|
||
items_in = items_in or []
|
||
for i, raw in enumerate(items_in):
|
||
itype = raw.get("item_type")
|
||
if not itype:
|
||
itype = "exercise" if raw.get("exercise_id") else "note"
|
||
order_ix = raw.get("order_index")
|
||
if order_ix is None:
|
||
order_ix = i
|
||
order_ix = int(order_ix)
|
||
|
||
if itype == "note":
|
||
body = raw.get("note_body")
|
||
if body is None:
|
||
body = ""
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_module_items (
|
||
module_id, order_index, item_type,
|
||
exercise_id, exercise_variant_id,
|
||
planned_duration_min, notes, note_body
|
||
) VALUES (%s, %s, 'note',
|
||
NULL, NULL, NULL, NULL, %s)
|
||
""",
|
||
(module_id, order_ix, body),
|
||
)
|
||
continue
|
||
|
||
eid = raw.get("exercise_id")
|
||
if not eid:
|
||
continue
|
||
eid = int(eid)
|
||
vid = _optional_positive_int(raw.get("exercise_variant_id"), "exercise_variant_id")
|
||
_validate_variant_for_exercise(cur, eid, vid)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_module_items (
|
||
module_id, order_index, item_type,
|
||
exercise_id, exercise_variant_id,
|
||
planned_duration_min, notes, note_body
|
||
) VALUES (%s, %s, 'exercise',
|
||
%s, %s, %s, %s, NULL)
|
||
""",
|
||
(
|
||
module_id,
|
||
order_ix,
|
||
eid,
|
||
vid,
|
||
raw.get("planned_duration_min"),
|
||
raw.get("notes"),
|
||
),
|
||
)
|
||
|
||
|
||
def load_training_module_for_apply(
|
||
cur, module_id: int, profile_id: int, role: Optional[str]
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
"""
|
||
Liest Modul inkl. Items für Übernahme in eine Trainingseinheit.
|
||
Returns (items_ordered, module_id).
|
||
Raises HTTPException bei 403/404.
|
||
"""
|
||
row = _fetch_training_module_row(cur, module_id)
|
||
_module_assert_readable(cur, row, profile_id, role or "")
|
||
cur.execute(
|
||
"""
|
||
SELECT item_type, exercise_id, exercise_variant_id,
|
||
planned_duration_min, notes, note_body
|
||
FROM training_module_items
|
||
WHERE module_id = %s
|
||
ORDER BY order_index ASC
|
||
""",
|
||
(module_id,),
|
||
)
|
||
raw_items = [r2d(x) for x in cur.fetchall()]
|
||
items: List[Dict[str, Any]] = []
|
||
for r in raw_items:
|
||
items.append(dict(r))
|
||
return items, int(module_id)
|
||
|
||
|
||
@router.get("/training-modules")
|
||
def list_training_modules(tenant: TenantContext = Depends(get_tenant_context)):
|
||
profile_id = tenant.profile_id
|
||
role = tenant.global_role
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
vis_clause, vis_params = library_content_visibility_sql(
|
||
alias="m",
|
||
profile_id=profile_id,
|
||
role=role,
|
||
effective_club_id=tenant.effective_club_id,
|
||
)
|
||
cur.execute(
|
||
f"""
|
||
SELECT m.*,
|
||
(SELECT COUNT(*) FROM training_module_items i WHERE i.module_id = m.id)
|
||
AS items_count
|
||
FROM training_modules m
|
||
WHERE ({vis_clause})
|
||
ORDER BY m.updated_at DESC NULLS LAST, m.title
|
||
""",
|
||
vis_params,
|
||
)
|
||
return [r2d(r) for r in cur.fetchall()]
|
||
|
||
|
||
@router.get("/training-modules/{module_id}")
|
||
def get_training_module(module_id: int, tenant: TenantContext = Depends(get_tenant_context)):
|
||
profile_id = tenant.profile_id
|
||
role = tenant.global_role
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
row = _module_access(cur, module_id, profile_id, role)
|
||
cur.execute(
|
||
"""
|
||
SELECT *
|
||
FROM training_module_items
|
||
WHERE module_id = %s
|
||
ORDER BY order_index ASC
|
||
""",
|
||
(module_id,),
|
||
)
|
||
row["items"] = [r2d(r) for r in cur.fetchall()]
|
||
return row
|
||
|
||
|
||
@router.post("/training-modules")
|
||
def create_training_module(data: dict, tenant: TenantContext = Depends(get_tenant_context)):
|
||
profile_id = tenant.profile_id
|
||
role = tenant.global_role
|
||
if not _has_planning_role(role):
|
||
raise HTTPException(status_code=403, detail="Nur Trainer dürfen Trainingsmodule anlegen")
|
||
|
||
title = (data.get("title") or "").strip()
|
||
if not title:
|
||
raise HTTPException(status_code=400, detail="title ist Pflicht")
|
||
|
||
vis_raw = data.get("visibility")
|
||
visibility = (vis_raw if isinstance(vis_raw, str) else "club").strip() or "club"
|
||
club_id = data.get("club_id")
|
||
if club_id in ("", []):
|
||
club_id = None
|
||
if visibility == "club" and club_id is None:
|
||
club_id = tenant.effective_club_id
|
||
|
||
primary_method_id = data.get("primary_method_id")
|
||
if primary_method_id in ("", []):
|
||
primary_method_id = None
|
||
if primary_method_id is not None:
|
||
primary_method_id = int(primary_method_id)
|
||
|
||
items_in = data.get("items") or []
|
||
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
assert_valid_governance_visibility(cur, profile_id, role, visibility, club_id)
|
||
if primary_method_id is not None:
|
||
cur.execute("SELECT 1 FROM training_methods WHERE id = %s", (primary_method_id,))
|
||
if not cur.fetchone():
|
||
raise HTTPException(status_code=400, detail="Trainingsmethode nicht gefunden")
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO training_modules (
|
||
club_id, created_by, title, summary, goal,
|
||
recommended_duration_min, target_group_notes, deployment_context_notes,
|
||
primary_method_id, visibility
|
||
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
RETURNING id
|
||
""",
|
||
(
|
||
club_id,
|
||
profile_id,
|
||
title,
|
||
(data.get("summary") or "").strip() or None,
|
||
data.get("goal"),
|
||
data.get("recommended_duration_min"),
|
||
data.get("target_group_notes"),
|
||
data.get("deployment_context_notes"),
|
||
primary_method_id,
|
||
visibility,
|
||
),
|
||
)
|
||
mid = cur.fetchone()["id"]
|
||
_replace_module_items(cur, mid, items_in if isinstance(items_in, list) else [])
|
||
conn.commit()
|
||
|
||
return get_training_module(mid, tenant)
|
||
|
||
|
||
@router.put("/training-modules/{module_id}")
|
||
def update_training_module(
|
||
module_id: int, data: dict, tenant: TenantContext = Depends(get_tenant_context)
|
||
):
|
||
profile_id = tenant.profile_id
|
||
role = tenant.global_role
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
row_prev = _fetch_training_module_row(cur, module_id)
|
||
assert_library_content_editable(cur, profile_id, role, row_prev)
|
||
|
||
merged_vis = row_prev.get("visibility") or "club"
|
||
merged_club = row_prev.get("club_id")
|
||
|
||
if "visibility" in data:
|
||
v_in = data.get("visibility")
|
||
if not isinstance(v_in, str) or v_in not in ("private", "club", "official"):
|
||
raise HTTPException(status_code=400, detail="visibility ungültig")
|
||
merged_vis = v_in
|
||
|
||
if "club_id" in data:
|
||
merged_club = data.get("club_id")
|
||
if merged_club in ("", []):
|
||
merged_club = None
|
||
if merged_vis == "club" and merged_club is None:
|
||
merged_club = tenant.effective_club_id
|
||
if "visibility" in data or "club_id" in data:
|
||
assert_library_content_governance_transition(
|
||
cur, profile_id, role, row_prev, merged_vis, merged_club
|
||
)
|
||
assert_valid_governance_visibility(cur, profile_id, role, merged_vis, merged_club)
|
||
|
||
fields: List[str] = []
|
||
params: List[Any] = []
|
||
|
||
if "title" in data:
|
||
t = data.get("title")
|
||
t = t.strip() if isinstance(t, str) else ""
|
||
if not t:
|
||
raise HTTPException(status_code=400, detail="title ist Pflicht")
|
||
fields.append("title = %s")
|
||
params.append(t)
|
||
|
||
for col in ("summary", "goal", "target_group_notes", "deployment_context_notes"):
|
||
if col in data:
|
||
fields.append(f"{col} = %s")
|
||
v = data.get(col)
|
||
if col == "summary" and isinstance(v, str):
|
||
v = v.strip() or None
|
||
params.append(v)
|
||
|
||
if "recommended_duration_min" in data:
|
||
fields.append("recommended_duration_min = %s")
|
||
params.append(data.get("recommended_duration_min"))
|
||
|
||
if "primary_method_id" in data:
|
||
pm = data.get("primary_method_id")
|
||
if pm in ("", [], None):
|
||
fields.append("primary_method_id = %s")
|
||
params.append(None)
|
||
else:
|
||
pm = int(pm)
|
||
cur.execute("SELECT 1 FROM training_methods WHERE id = %s", (pm,))
|
||
if not cur.fetchone():
|
||
raise HTTPException(status_code=400, detail="Trainingsmethode nicht gefunden")
|
||
fields.append("primary_method_id = %s")
|
||
params.append(pm)
|
||
|
||
if "club_id" in data:
|
||
fields.append("club_id = %s")
|
||
params.append(merged_club)
|
||
|
||
if "visibility" in data:
|
||
fields.append("visibility = %s")
|
||
params.append(merged_vis)
|
||
|
||
if fields:
|
||
fields.append("updated_at = NOW()")
|
||
params.append(module_id)
|
||
cur.execute(
|
||
f"UPDATE training_modules SET {', '.join(fields)} WHERE id = %s",
|
||
tuple(params),
|
||
)
|
||
|
||
if "items" in data:
|
||
items_in = data["items"]
|
||
_replace_module_items(cur, module_id, items_in if isinstance(items_in, list) else [])
|
||
|
||
conn.commit()
|
||
|
||
return get_training_module(module_id, tenant)
|
||
|
||
|
||
@router.delete("/training-modules/{module_id}")
|
||
def delete_training_module(module_id: int, tenant: TenantContext = Depends(get_tenant_context)):
|
||
profile_id = tenant.profile_id
|
||
role = tenant.global_role
|
||
with get_db() as conn:
|
||
cur = get_cursor(conn)
|
||
row_del = _fetch_training_module_row(cur, module_id)
|
||
assert_library_content_deletable(cur, profile_id, role, row_del)
|
||
cur.execute("DELETE FROM training_modules WHERE id = %s", (module_id,))
|
||
conn.commit()
|
||
return {"ok": True}
|