shinkan-jinkendo/backend/routers/maturity_models.py
Lars f1ee1eec7e
Some checks failed
Deploy Development / deploy (push) Successful in 35s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 5s
Test Suite / playwright-tests (push) Failing after 1m55s
feat: update maturity models and version to 0.7.2
- Incremented application version to 0.7.2 and updated database schema version to 20260427025.
- Enhanced maturity models functionality to support M:N relationships for focus areas, style directions, and target groups.
- Updated frontend to allow multi-selection for focus areas, style directions, and target groups.
- Documented changes in the changelog for version 0.7.2, including new migration details and UI improvements.
2026-04-27 11:42:03 +02:00

606 lines
21 KiB
Python

"""
Reifegradmodelle / Fähigkeitsmatrix (kontextbezogen)
Kontext zu Fokusbereich, Stilrichtung, Zielgruppe: jeweils M:N (leer = gilt überall).
Lesen: alle authentifizierten Nutzer.
Schreiben: admin, superadmin.
"""
from typing import Any, Dict, List, Optional, Sequence
from fastapi import APIRouter, Depends, HTTPException, Query
from auth import require_auth
from db import get_db, get_cursor, r2d
router = APIRouter(prefix="/api", tags=["maturity_models"])
def _require_admin(session: dict) -> None:
role = session.get("role")
if role not in ("admin", "superadmin"):
raise HTTPException(403, "Nur Administratoren dürfen Reifegradmodelle verwalten")
def _base_maturity_model(cur, model_id: int) -> Optional[Dict[str, Any]]:
cur.execute("SELECT * FROM maturity_models WHERE id = %s", (model_id,))
row = cur.fetchone()
return r2d(row) if row else None
def _attach_context(cur, m: Dict[str, Any]) -> Dict[str, Any]:
mid = m["id"]
cur.execute(
"""
SELECT fa.id, fa.name, fa.abbreviation, mfa.is_primary
FROM maturity_model_focus_areas mfa
JOIN focus_areas fa ON fa.id = mfa.focus_area_id
WHERE mfa.maturity_model_id = %s
ORDER BY mfa.is_primary DESC NULLS LAST, fa.sort_order, fa.name
""",
(mid,),
)
m["focus_areas"] = [r2d(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT sd.id, sd.name, msd.is_primary
FROM maturity_model_style_directions msd
JOIN style_directions sd ON sd.id = msd.style_direction_id
WHERE msd.maturity_model_id = %s
ORDER BY msd.is_primary DESC NULLS LAST, sd.name
""",
(mid,),
)
m["style_directions"] = [r2d(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT tg.id, tg.name, mtg.is_primary
FROM maturity_model_target_groups mtg
JOIN target_groups tg ON tg.id = mtg.target_group_id
WHERE mtg.maturity_model_id = %s
ORDER BY mtg.is_primary DESC NULLS LAST, tg.name
""",
(mid,),
)
m["target_groups"] = [r2d(r) for r in cur.fetchall()]
m["focus_area_name"] = m["focus_areas"][0]["name"] if m["focus_areas"] else None
m["style_direction_name"] = m["style_directions"][0]["name"] if m["style_directions"] else None
m["target_group_name"] = m["target_groups"][0]["name"] if m["target_groups"] else None
return m
def _normalize_id_list(raw: Any) -> Optional[List[int]]:
if raw is None:
return None
if isinstance(raw, (str, bytes)):
raise HTTPException(400, "Erwarte Liste von IDs")
if not isinstance(raw, Sequence):
raise HTTPException(400, "Erwarte Liste von IDs")
out: List[int] = []
for x in raw:
if x is None:
continue
out.append(int(x))
return out
def _write_context_junctions(cur, model_id: int, data: Dict[str, Any]) -> None:
fa = data.get("focus_area_ids")
if fa is None and data.get("focus_area_id") is not None:
fa = [data.get("focus_area_id")]
if fa is not None:
fa = _normalize_id_list(fa)
cur.execute("DELETE FROM maturity_model_focus_areas WHERE maturity_model_id = %s", (model_id,))
for i, fid in enumerate(fa or []):
cur.execute(
"""
INSERT INTO maturity_model_focus_areas (maturity_model_id, focus_area_id, is_primary)
VALUES (%s, %s, %s)
""",
(model_id, fid, i == 0),
)
sd = data.get("style_direction_ids")
if sd is None and data.get("style_direction_id") is not None:
sd = [data.get("style_direction_id")]
if sd is not None:
sd = _normalize_id_list(sd)
cur.execute("DELETE FROM maturity_model_style_directions WHERE maturity_model_id = %s", (model_id,))
for i, sid in enumerate(sd or []):
cur.execute(
"""
INSERT INTO maturity_model_style_directions (maturity_model_id, style_direction_id, is_primary)
VALUES (%s, %s, %s)
""",
(model_id, sid, i == 0),
)
tg = data.get("target_group_ids")
if tg is None and data.get("target_group_id") is not None:
tg = [data.get("target_group_id")]
if tg is not None:
tg = _normalize_id_list(tg)
cur.execute("DELETE FROM maturity_model_target_groups WHERE maturity_model_id = %s", (model_id,))
for i, tid in enumerate(tg or []):
cur.execute(
"""
INSERT INTO maturity_model_target_groups (maturity_model_id, target_group_id, is_primary)
VALUES (%s, %s, %s)
""",
(model_id, tid, i == 0),
)
def _load_full_model(cur, model_id: int) -> Dict[str, Any]:
base = _base_maturity_model(cur, model_id)
if not base:
raise HTTPException(404, "Reifegradmodell nicht gefunden")
_attach_context(cur, base)
cur.execute(
"""
SELECT * FROM model_levels
WHERE maturity_model_id = %s
ORDER BY sort_order ASC, level_number ASC
""",
(model_id,),
)
levels = [r2d(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT ms.*, s.name AS skill_name, s.status AS skill_status,
sc.name AS skill_subcategory_name, sc.slug AS skill_subcategory_slug,
mc.name AS skill_main_category_name, mc.slug AS skill_main_category_slug
FROM model_skills ms
JOIN skills s ON s.id = ms.skill_id
LEFT JOIN skill_categories sc ON s.category_id = sc.id
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
WHERE ms.maturity_model_id = %s
ORDER BY ms.sort_order ASC, mc.sort_order NULLS LAST, sc.sort_order NULLS LAST, s.name ASC
""",
(model_id,),
)
model_skills = [r2d(r) for r in cur.fetchall()]
cur.execute(
"""
SELECT msl.*, s.name AS skill_name
FROM model_skill_levels msl
JOIN skills s ON s.id = msl.skill_id
WHERE msl.maturity_model_id = %s
ORDER BY s.name ASC, msl.level_number ASC
""",
(model_id,),
)
skill_levels = [r2d(r) for r in cur.fetchall()]
return {
**base,
"levels": levels,
"model_skills": model_skills,
"skill_levels": skill_levels,
}
def _insert_default_levels(cur, model_id: int, level_count: int) -> None:
for i in range(1, level_count + 1):
cur.execute(
"""
INSERT INTO model_levels (maturity_model_id, level_number, name, description, sort_order)
VALUES (%s, %s, %s, %s, %s)
""",
(model_id, i, f"Stufe {i}", None, i),
)
def _replace_levels(cur, model_id: int, level_count: int, levels: List[Dict[str, Any]]) -> None:
if len(levels) != level_count:
raise HTTPException(
400,
f"Anzahl der Stufen-Definitionen ({len(levels)}) muss level_count ({level_count}) entsprechen",
)
seen = set()
for lev in levels:
num = int(lev.get("level_number") or 0)
if num < 1 or num > level_count:
raise HTTPException(400, f"Ungültige level_number: {num}")
if num in seen:
raise HTTPException(400, f"Doppelte level_number: {num}")
seen.add(num)
if seen != set(range(1, level_count + 1)):
raise HTTPException(400, "level_number muss lückenlos 1..level_count abdecken")
cur.execute("DELETE FROM model_levels WHERE maturity_model_id = %s", (model_id,))
for lev in sorted(levels, key=lambda x: int(x.get("level_number"))):
cur.execute(
"""
INSERT INTO model_levels (maturity_model_id, level_number, name, description, sort_order)
VALUES (%s, %s, %s, %s, %s)
""",
(
model_id,
int(lev["level_number"]),
(lev.get("name") or "").strip() or f"Stufe {lev['level_number']}",
lev.get("description"),
int(lev.get("sort_order") or lev["level_number"]),
),
)
def _dim_matches(items: List[Dict[str, Any]], query_id: Optional[int]) -> bool:
if not items:
return True
if query_id is None:
return True
return any(int(x["id"]) == int(query_id) for x in items)
def _dim_score(items: List[Dict[str, Any]], query_id: Optional[int]) -> int:
if not items:
return 0
if query_id is not None and any(int(x["id"]) == int(query_id) for x in items):
return 1
return 0
@router.get("/maturity-models/resolve")
def resolve_maturity_model(
focus_area_id: Optional[int] = Query(default=None),
style_direction_id: Optional[int] = Query(default=None),
target_group_id: Optional[int] = Query(default=None),
session: dict = Depends(require_auth),
):
"""
Wählt das spezifischste aktive Modell, das zum Kontext passt.
Leere M:N-Zuordnung = Wildcard für diese Dimension.
"""
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT * FROM maturity_models WHERE status = 'active' ORDER BY id")
rows = [r2d(r) for r in cur.fetchall()]
enriched: List[Dict[str, Any]] = []
with get_db() as conn:
cur = get_cursor(conn)
for m in rows:
_attach_context(cur, m)
enriched.append(m)
def ok(m: Dict[str, Any]) -> bool:
if not _dim_matches(m.get("focus_areas") or [], focus_area_id):
return False
if not _dim_matches(m.get("style_directions") or [], style_direction_id):
return False
if not _dim_matches(m.get("target_groups") or [], target_group_id):
return False
return True
def score(m: Dict[str, Any]) -> int:
return (
_dim_score(m.get("focus_areas") or [], focus_area_id)
+ _dim_score(m.get("style_directions") or [], style_direction_id)
+ _dim_score(m.get("target_groups") or [], target_group_id)
)
candidates = [m for m in enriched if ok(m)]
if not candidates:
return None
best = max(candidates, key=lambda m: (score(m), m.get("id") or 0))
model_id = best["id"]
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)
@router.get("/maturity-models")
def list_maturity_models(
status: Optional[str] = Query(default=None),
focus_area_id: Optional[int] = Query(default=None),
session: dict = Depends(require_auth),
):
with get_db() as conn:
cur = get_cursor(conn)
q = "SELECT * FROM maturity_models mm WHERE 1=1"
params: List[Any] = []
if status:
q += " AND mm.status = %s"
params.append(status)
if focus_area_id is not None:
q += """ AND (
EXISTS (
SELECT 1 FROM maturity_model_focus_areas mfa
WHERE mfa.maturity_model_id = mm.id
AND mfa.focus_area_id = %s
)
OR NOT EXISTS (
SELECT 1 FROM maturity_model_focus_areas mfa2
WHERE mfa2.maturity_model_id = mm.id
)
)"""
params.append(focus_area_id)
q += " ORDER BY mm.name ASC"
cur.execute(q, params)
rows = [r2d(r) for r in cur.fetchall()]
for m in rows:
_attach_context(cur, m)
return rows
@router.get("/maturity-models/{model_id}")
def get_maturity_model(model_id: int, session: dict = Depends(require_auth)):
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)
@router.post("/maturity-models")
def create_maturity_model(data: Dict[str, Any], session: dict = Depends(require_auth)):
_require_admin(session)
name = (data.get("name") or "").strip()
if not name:
raise HTTPException(400, "Name ist Pflichtfeld")
level_count = int(data.get("level_count") or 5)
if level_count < 3 or level_count > 10:
raise HTTPException(400, "level_count muss zwischen 3 und 10 liegen")
profile_id = session.get("profile_id")
with get_db() as conn:
cur = get_cursor(conn)
try:
cur.execute(
"""
INSERT INTO maturity_models (
name, description,
level_count, status, version,
created_by, club_id,
import_source, import_id
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
RETURNING id
""",
(
name,
data.get("description"),
level_count,
data.get("status") or "draft",
data.get("version") or "1.0",
profile_id,
data.get("club_id"),
data.get("import_source"),
data.get("import_id"),
),
)
mid = cur.fetchone()["id"]
except Exception as e:
if "unique" in str(e).lower() or "duplicate" in str(e).lower():
raise HTTPException(409, "Konflikt: import_id oder Name bereits vergeben") from e
raise
if any(
k in data
for k in (
"focus_area_ids",
"style_direction_ids",
"target_group_ids",
"focus_area_id",
"style_direction_id",
"target_group_id",
)
):
_write_context_junctions(cur, mid, data)
levels = data.get("levels")
if levels:
_replace_levels(cur, mid, level_count, levels)
else:
_insert_default_levels(cur, mid, level_count)
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, mid)
@router.put("/maturity-models/{model_id}")
def update_maturity_model(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
_require_admin(session)
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT * FROM maturity_models WHERE id = %s", (model_id,))
row = cur.fetchone()
if not row:
raise HTTPException(404, "Reifegradmodell nicht gefunden")
current = r2d(row)
level_count = int(current["level_count"])
if "level_count" in data and data["level_count"] is not None:
level_count = int(data["level_count"])
if level_count < 3 or level_count > 10:
raise HTTPException(400, "level_count muss zwischen 3 und 10 liegen")
if level_count != int(current["level_count"]):
if "levels" not in data or data["levels"] is None:
raise HTTPException(
400,
"Stufenanzahl ändern nur zusammen mit vollständiger levels-Liste (Stufen-Editor).",
)
if level_count < int(current["level_count"]):
cur.execute(
"DELETE FROM model_skill_levels WHERE maturity_model_id = %s AND level_number > %s",
(model_id, level_count),
)
sets: List[str] = []
vals: List[Any] = []
for key in ("name", "description", "status", "version"):
if key in data:
sets.append(f"{key} = %s")
vals.append(data[key])
sets.append("level_count = %s")
vals.append(level_count)
sets.append("updated_at = NOW()")
vals.append(model_id)
cur.execute(
f"UPDATE maturity_models SET {', '.join(sets)} WHERE id = %s",
tuple(vals),
)
if any(
k in data
for k in (
"focus_area_ids",
"style_direction_ids",
"target_group_ids",
"focus_area_id",
"style_direction_id",
"target_group_id",
)
):
_write_context_junctions(cur, model_id, data)
if "levels" in data and data["levels"] is not None:
_replace_levels(cur, model_id, level_count, data["levels"])
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)
@router.delete("/maturity-models/{model_id}")
def delete_maturity_model(model_id: int, session: dict = Depends(require_auth)):
role = session.get("role")
if role != "superadmin":
raise HTTPException(403, "Nur Superadmins dürfen Reifegradmodelle löschen")
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("DELETE FROM maturity_models WHERE id = %s RETURNING id", (model_id,))
if not cur.fetchone():
raise HTTPException(404, "Reifegradmodell nicht gefunden")
return {"ok": True}
@router.post("/maturity-models/{model_id}/skills")
def add_model_skill(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
_require_admin(session)
skill_id = data.get("skill_id")
if not skill_id:
raise HTTPException(400, "skill_id ist Pflicht")
with get_db() as conn:
cur = get_cursor(conn)
if not _base_maturity_model(cur, model_id):
raise HTTPException(404, "Reifegradmodell nicht gefunden")
cur.execute("SELECT id FROM skills WHERE id = %s", (skill_id,))
if not cur.fetchone():
raise HTTPException(404, "Fähigkeit nicht gefunden")
try:
cur.execute(
"""
INSERT INTO model_skills (maturity_model_id, skill_id, sort_order, relevance)
VALUES (%s, %s, %s, %s)
RETURNING id
""",
(
model_id,
skill_id,
int(data.get("sort_order") or 0),
data.get("relevance"),
),
)
except Exception as e:
if "unique" in str(e).lower():
raise HTTPException(409, "Fähigkeit ist bereits im Modell") from e
raise
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)
@router.delete("/maturity-models/{model_id}/skills/{skill_id}")
def remove_model_skill(model_id: int, skill_id: int, session: dict = Depends(require_auth)):
_require_admin(session)
with get_db() as conn:
cur = get_cursor(conn)
cur.execute(
"DELETE FROM model_skills WHERE maturity_model_id = %s AND skill_id = %s RETURNING id",
(model_id, skill_id),
)
if not cur.fetchone():
raise HTTPException(404, "Zuordnung nicht gefunden")
cur.execute(
"DELETE FROM model_skill_levels WHERE maturity_model_id = %s AND skill_id = %s",
(model_id, skill_id),
)
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)
@router.put("/maturity-models/{model_id}/skill-levels")
def upsert_model_skill_levels(model_id: int, data: Dict[str, Any], session: dict = Depends(require_auth)):
_require_admin(session)
entries: List[Dict[str, Any]] = data.get("entries") or []
if not entries:
raise HTTPException(400, "entries darf nicht leer sein")
with get_db() as conn:
cur = get_cursor(conn)
cur.execute("SELECT level_count FROM maturity_models WHERE id = %s", (model_id,))
row = cur.fetchone()
if not row:
raise HTTPException(404, "Reifegradmodell nicht gefunden")
level_count = int(row["level_count"])
for e in entries:
sid = e.get("skill_id")
ln = int(e.get("level_number") or 0)
desc = (e.get("description") or "").strip()
if not sid or ln < 1 or ln > level_count:
raise HTTPException(400, "Ungültige skill_id oder level_number")
if not desc:
cur.execute(
"""
DELETE FROM model_skill_levels
WHERE maturity_model_id = %s AND skill_id = %s AND level_number = %s
""",
(model_id, sid, ln),
)
continue
cur.execute(
"""
INSERT INTO model_skill_levels (
maturity_model_id, skill_id, level_number,
description, observable_criteria, example_exercise_hints
) VALUES (%s,%s,%s,%s,%s,%s)
ON CONFLICT (maturity_model_id, skill_id, level_number)
DO UPDATE SET
description = EXCLUDED.description,
observable_criteria = EXCLUDED.observable_criteria,
example_exercise_hints = EXCLUDED.example_exercise_hints,
updated_at = NOW()
""",
(
model_id,
sid,
ln,
desc,
e.get("observable_criteria"),
e.get("example_exercise_hints"),
),
)
with get_db() as conn:
cur = get_cursor(conn)
return _load_full_model(cur, model_id)