Lars
3397b2094d
- Incremented application version to 0.7.3 and updated database schema version to 20260427026. - Enhanced maturity model context bindings with new hierarchical resolution logic for focus areas, style directions, and training types. - Added new API endpoints for managing maturity model context bindings. - Updated frontend components to support the new context binding functionality and improved admin UI for better user experience. - Documented changes in the changelog for version 0.7.3, including new features and improvements.
900 lines
32 KiB
Python
900 lines
32 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
|
|
|
|
|
|
def _binding_model_active(cur, maturity_model_id: int) -> bool:
|
|
b = _base_maturity_model(cur, maturity_model_id)
|
|
return bool(b and b.get("status") == "active")
|
|
|
|
|
|
def _resolve_binding_model_ids(
|
|
cur,
|
|
focus_area_id: int,
|
|
style_direction_id: Optional[int],
|
|
training_type_id: Optional[int],
|
|
) -> List[int]:
|
|
"""
|
|
Kette Basis → Stilrichtung → Trainingsstil (training_types).
|
|
Ohne Basis-Zeile (nur Fokus): leere Liste → Aufrufer nutzt Legacy-Resolve.
|
|
Nur **aktive** Modelle werden eingereiht; inaktive Overlays werden übersprungen.
|
|
"""
|
|
chain: List[int] = []
|
|
cur.execute(
|
|
"""
|
|
SELECT maturity_model_id FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id IS NULL
|
|
""",
|
|
(focus_area_id,),
|
|
)
|
|
r = cur.fetchone()
|
|
if not r:
|
|
return []
|
|
mid0 = int(r["maturity_model_id"])
|
|
if not _binding_model_active(cur, mid0):
|
|
return []
|
|
chain.append(mid0)
|
|
|
|
if style_direction_id is not None:
|
|
cur.execute(
|
|
"""
|
|
SELECT maturity_model_id FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id IS NULL
|
|
""",
|
|
(focus_area_id, style_direction_id),
|
|
)
|
|
r2 = cur.fetchone()
|
|
if r2 and _binding_model_active(cur, int(r2["maturity_model_id"])):
|
|
chain.append(int(r2["maturity_model_id"]))
|
|
|
|
if style_direction_id is not None and training_type_id is not None:
|
|
cur.execute(
|
|
"""
|
|
SELECT maturity_model_id FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id = %s
|
|
""",
|
|
(focus_area_id, style_direction_id, training_type_id),
|
|
)
|
|
r3 = cur.fetchone()
|
|
if r3 and _binding_model_active(cur, int(r3["maturity_model_id"])):
|
|
chain.append(int(r3["maturity_model_id"]))
|
|
|
|
return chain
|
|
|
|
|
|
def _merge_loaded_models(loaded: List[Dict[str, Any]]) -> Dict[str, Any]:
|
|
"""Überlagert Zelltexte in Reihenfolge der Kette; Zeilen kommen aus Basis + fehlende Fähigkeiten aus Overlays."""
|
|
if not loaded:
|
|
raise ValueError("merge: keine Modelle")
|
|
base = loaded[0]
|
|
base_lc = int(base.get("level_count") or 5)
|
|
base_levels = base.get("levels") or []
|
|
|
|
skill_rows: Dict[int, Dict[str, Any]] = {}
|
|
order: List[int] = []
|
|
for ms in base.get("model_skills") or []:
|
|
sid = int(ms["skill_id"])
|
|
skill_rows[sid] = dict(ms)
|
|
order.append(sid)
|
|
|
|
for fm in loaded[1:]:
|
|
for ms in fm.get("model_skills") or []:
|
|
sid = int(ms["skill_id"])
|
|
if sid not in skill_rows:
|
|
row = dict(ms)
|
|
row["maturity_model_id"] = base["id"]
|
|
skill_rows[sid] = row
|
|
order.append(sid)
|
|
|
|
merged_model_skills = [skill_rows[sid] for sid in order]
|
|
|
|
cell_map: Dict[tuple, Dict[str, Any]] = {}
|
|
for fm in loaded:
|
|
for sl in fm.get("skill_levels") or []:
|
|
ln = int(sl["level_number"])
|
|
if ln < 1 or ln > base_lc:
|
|
continue
|
|
sid = int(sl["skill_id"])
|
|
key = (sid, ln)
|
|
cell_map[key] = {
|
|
"maturity_model_id": base["id"],
|
|
"skill_id": sid,
|
|
"level_number": ln,
|
|
"description": sl.get("description") or "",
|
|
"observable_criteria": sl.get("observable_criteria"),
|
|
"example_exercise_hints": sl.get("example_exercise_hints"),
|
|
"ai_generated": sl.get("ai_generated"),
|
|
"skill_name": sl.get("skill_name"),
|
|
}
|
|
|
|
for sid, ln in list(cell_map.keys()):
|
|
row = cell_map[(sid, ln)]
|
|
if not row.get("skill_name"):
|
|
row["skill_name"] = skill_rows.get(sid, {}).get("skill_name")
|
|
|
|
merged_skill_levels = list(cell_map.values())
|
|
merged_skill_levels.sort(key=lambda x: (x.get("skill_name") or "", x["level_number"]))
|
|
|
|
out = {
|
|
**base,
|
|
"model_skills": merged_model_skills,
|
|
"skill_levels": merged_skill_levels,
|
|
"levels": base_levels,
|
|
"level_count": base_lc,
|
|
"resolution": {
|
|
"merged": len(loaded) > 1,
|
|
"source_model_ids": [int(m["id"]) for m in loaded],
|
|
"binding_strategy": "hierarchical_override",
|
|
},
|
|
}
|
|
return out
|
|
|
|
|
|
def _legacy_resolve_pick_model_id(
|
|
cur,
|
|
focus_area_id: Optional[int],
|
|
style_direction_id: Optional[int],
|
|
target_group_id: Optional[int],
|
|
) -> Optional[int]:
|
|
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]] = []
|
|
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))
|
|
return int(best["id"])
|
|
|
|
|
|
@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),
|
|
training_type_id: Optional[int] = Query(
|
|
default=None,
|
|
description="Trainingsstil (training_types, z. B. Leistungssport); nur mit Stilrichtung sinnvoll.",
|
|
),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Liefert die Fähigkeitsmatrix zum Kontext.
|
|
|
|
**Priorität 1:** Tabelle `maturity_model_context_bindings` (Fokus → optional Stilrichtung → optional Trainingsstil).
|
|
Mehrere Modelle werden zusammengeführt: spätere Stufen überschreiben Zelltexte für dieselbe Fähigkeit/Stufe.
|
|
|
|
**Priorität 2 (Legacy):** Ein aktives Modell, dessen M:N-Kontext zum Filter passt (Zielgruppe unverändert).
|
|
"""
|
|
if focus_area_id is not None:
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
chain = _resolve_binding_model_ids(
|
|
cur,
|
|
int(focus_area_id),
|
|
style_direction_id,
|
|
training_type_id,
|
|
)
|
|
if chain:
|
|
loaded = [_load_full_model(cur, mid) for mid in chain]
|
|
return _merge_loaded_models(loaded)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
mid = _legacy_resolve_pick_model_id(
|
|
cur, focus_area_id, style_direction_id, target_group_id
|
|
)
|
|
if mid is None:
|
|
return None
|
|
return _load_full_model(cur, mid)
|
|
|
|
|
|
@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)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
# Kontext-Bindings (Fokus / Stilrichtung / Trainingsstil) — hierarchisch
|
|
# ═══════════════════════════════════════════════════════════════════════════
|
|
|
|
|
|
@router.get("/maturity-model-context-bindings")
|
|
def list_maturity_model_context_bindings(session: dict = Depends(require_auth)):
|
|
role = session.get("role")
|
|
if role not in ("admin", "superadmin"):
|
|
raise HTTPException(403, "Nur Administratoren")
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT b.id, b.maturity_model_id, mm.name AS maturity_model_name, mm.status AS model_status,
|
|
b.focus_area_id, fa.name AS focus_area_name,
|
|
b.style_direction_id, sd.name AS style_direction_name,
|
|
b.training_type_id, tt.name AS training_type_name
|
|
FROM maturity_model_context_bindings b
|
|
JOIN maturity_models mm ON mm.id = b.maturity_model_id
|
|
JOIN focus_areas fa ON fa.id = b.focus_area_id
|
|
LEFT JOIN style_directions sd ON sd.id = b.style_direction_id
|
|
LEFT JOIN training_types tt ON tt.id = b.training_type_id
|
|
ORDER BY fa.sort_order NULLS LAST, fa.name,
|
|
(b.style_direction_id IS NULL) DESC,
|
|
sd.name NULLS LAST,
|
|
(b.training_type_id IS NULL) DESC,
|
|
tt.name NULLS LAST
|
|
"""
|
|
)
|
|
return [r2d(r) for r in cur.fetchall()]
|
|
|
|
|
|
@router.post("/maturity-model-context-bindings")
|
|
def upsert_maturity_model_context_binding(data: Dict[str, Any], session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
mid = data.get("maturity_model_id")
|
|
fa = data.get("focus_area_id")
|
|
if mid is None or fa is None:
|
|
raise HTTPException(400, "maturity_model_id und focus_area_id sind Pflichtfelder")
|
|
mid = int(mid)
|
|
fa = int(fa)
|
|
sd_raw = data.get("style_direction_id")
|
|
tt_raw = data.get("training_type_id")
|
|
sd: Optional[int] = int(sd_raw) if sd_raw is not None else None
|
|
tt: Optional[int] = int(tt_raw) if tt_raw is not None else None
|
|
|
|
if sd is None and tt is not None:
|
|
raise HTTPException(400, "Trainingsstil nur zusammen mit Stilrichtung erlaubt")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if not _base_maturity_model(cur, mid):
|
|
raise HTTPException(404, "Reifegradmodell nicht gefunden")
|
|
cur.execute("SELECT id FROM focus_areas WHERE id = %s", (fa,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Fokusbereich nicht gefunden")
|
|
if sd is not None:
|
|
cur.execute("SELECT id FROM style_directions WHERE id = %s", (sd,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Stilrichtung nicht gefunden")
|
|
if tt is not None:
|
|
cur.execute("SELECT id FROM training_types WHERE id = %s", (tt,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Trainingsstil nicht gefunden")
|
|
|
|
if sd is None and tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id IS NULL AND training_type_id IS NULL
|
|
""",
|
|
(fa,),
|
|
)
|
|
elif tt is None:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id IS NULL
|
|
""",
|
|
(fa, sd),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
DELETE FROM maturity_model_context_bindings
|
|
WHERE focus_area_id = %s AND style_direction_id = %s AND training_type_id = %s
|
|
""",
|
|
(fa, sd, tt),
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO maturity_model_context_bindings (
|
|
maturity_model_id, focus_area_id, style_direction_id, training_type_id
|
|
)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(mid, fa, sd, tt),
|
|
)
|
|
new_id = cur.fetchone()["id"]
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT b.id, b.maturity_model_id, mm.name AS maturity_model_name, mm.status AS model_status,
|
|
b.focus_area_id, fa.name AS focus_area_name,
|
|
b.style_direction_id, sd.name AS style_direction_name,
|
|
b.training_type_id, tt.name AS training_type_name
|
|
FROM maturity_model_context_bindings b
|
|
JOIN maturity_models mm ON mm.id = b.maturity_model_id
|
|
JOIN focus_areas fa ON fa.id = b.focus_area_id
|
|
LEFT JOIN style_directions sd ON sd.id = b.style_direction_id
|
|
LEFT JOIN training_types tt ON tt.id = b.training_type_id
|
|
WHERE b.id = %s
|
|
""",
|
|
(new_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return r2d(row) if row else {"id": new_id}
|
|
|
|
|
|
@router.delete("/maturity-model-context-bindings/{binding_id}")
|
|
def delete_maturity_model_context_binding(binding_id: int, session: dict = Depends(require_auth)):
|
|
_require_admin(session)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM maturity_model_context_bindings WHERE id = %s RETURNING id",
|
|
(binding_id,),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Zuordnung nicht gefunden")
|
|
return {"ok": True}
|