Lars
949a77fe38
All checks were successful
Deploy Development / deploy (push) Successful in 43s
Test Suite / pytest-backend (push) Successful in 38s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 15s
Test Suite / k6 /health Baseline (push) Successful in 33s
Test Suite / playwright-tests (push) Successful in 1m11s
- Added `karate_relevance` and `relevance_level` fields to the SkillCreate and SkillResponse models, allowing for more detailed skill attributes. - Updated the SMW property mapping to include these new fields, facilitating their integration during data import. - Implemented parsing logic for relevance levels from Wiki data, ensuring proper handling of values between 1 and 3. - Modified the upsert and create skill functions to support the new fields, ensuring they are correctly stored and updated in the database. - Incremented app version to 0.8.143 and updated changelog to reflect these changes.
462 lines
15 KiB
Python
462 lines
15 KiB
Python
"""
|
|
Skills & Methods Catalog Endpoints for Shinkan Jinkendo
|
|
|
|
Handles CRUD operations for skills and training methods.
|
|
Read access for all authenticated users.
|
|
Write access for admins only.
|
|
"""
|
|
import json
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Query
|
|
from psycopg2.extras import Json
|
|
|
|
from db import get_db, get_cursor, r2d
|
|
from auth import require_auth
|
|
|
|
router = APIRouter(prefix="/api", tags=["skills"])
|
|
|
|
|
|
def _jsonb_param(val: Any) -> Any:
|
|
"""psycopg2 benötigt Json() oder String für JSONB — rohe dict/list sonst ProgrammingError."""
|
|
if val is None:
|
|
return None
|
|
if isinstance(val, (dict, list)):
|
|
return Json(val)
|
|
return val
|
|
|
|
|
|
def _main_id_for_category(cur, category_id: Optional[int]) -> Optional[int]:
|
|
if not category_id:
|
|
return None
|
|
cur.execute(
|
|
"SELECT main_category_id FROM skill_categories WHERE id = %s",
|
|
(category_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return row["main_category_id"] if row else None
|
|
|
|
|
|
# ── List Skills ───────────────────────────────────────────────────────
|
|
@router.get("/skills")
|
|
def list_skills(
|
|
category: Optional[str] = Query(default=None),
|
|
status: Optional[str] = Query(default=None),
|
|
session=Depends(require_auth)
|
|
):
|
|
"""
|
|
List all skills (public for authenticated users).
|
|
|
|
Filters:
|
|
- category: kihon, kumite, kata, selbstverteidigung, fitness
|
|
- status: active, inactive
|
|
"""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
query = "SELECT * FROM skills"
|
|
params = []
|
|
where = []
|
|
|
|
if category:
|
|
where.append("category = %s")
|
|
params.append(category)
|
|
|
|
if status:
|
|
if status == "active":
|
|
where.append("(status = 'active' OR status IS NULL)")
|
|
else:
|
|
where.append("status = %s")
|
|
params.append(status)
|
|
else:
|
|
# Default: nur aktive (NULL = Legacy, wie Kataloglisten)
|
|
where.append("(status = 'active' OR status IS NULL)")
|
|
|
|
if where:
|
|
query += " WHERE " + " AND ".join(where)
|
|
|
|
query += " ORDER BY importance DESC, name"
|
|
|
|
cur.execute(query, params)
|
|
rows = cur.fetchall()
|
|
return [r2d(r) for r in rows]
|
|
|
|
|
|
# ── Katalog-Ansicht (Hierarchie-Sortierung, Admin) ───────────────────────
|
|
@router.get("/skills/catalog")
|
|
def list_skills_catalog(
|
|
status: Optional[str] = Query(
|
|
default="active",
|
|
description="'active', 'inactive' oder 'all' (nur Admin)",
|
|
),
|
|
session=Depends(require_auth),
|
|
):
|
|
"""
|
|
Fähigkeiten für Admin-Katalog: sortiert nach Hauptgruppe → Kategorie → Sortierung → Name.
|
|
status=all nur für admin/superadmin.
|
|
"""
|
|
role = session.get("role")
|
|
admin = role in ("admin", "superadmin")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
q = """
|
|
SELECT s.*,
|
|
mc.name AS catalog_main_category_name,
|
|
mc.sort_order AS catalog_main_sort,
|
|
mc.slug AS catalog_main_slug,
|
|
sc.name AS catalog_category_name,
|
|
sc.sort_order AS catalog_category_sort,
|
|
sc.slug AS catalog_category_slug
|
|
FROM skills s
|
|
LEFT JOIN skill_main_categories mc ON s.main_category_id = mc.id
|
|
LEFT JOIN skill_categories sc ON s.category_id = sc.id
|
|
WHERE 1=1
|
|
"""
|
|
params: list[Any] = []
|
|
if status == "all":
|
|
if not admin:
|
|
raise HTTPException(403, "Nur Admins dürfen alle Status sehen")
|
|
elif status:
|
|
q += " AND s.status = %s"
|
|
params.append(status)
|
|
else:
|
|
q += " AND s.status = %s"
|
|
params.append("active")
|
|
q += """
|
|
ORDER BY mc.sort_order NULLS LAST,
|
|
sc.sort_order NULLS LAST,
|
|
s.sort_order NULLS LAST,
|
|
s.name
|
|
"""
|
|
cur.execute(q, params)
|
|
return [r2d(r) for r in cur.fetchall()]
|
|
|
|
|
|
# ── Get Skill ─────────────────────────────────────────────────────────
|
|
@router.get("/skills/{skill_id}")
|
|
def get_skill(skill_id: int, session=Depends(require_auth)):
|
|
"""Get skill by ID."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
cur.execute("SELECT * FROM skills WHERE id = %s", (skill_id,))
|
|
skill = cur.fetchone()
|
|
|
|
if not skill:
|
|
raise HTTPException(404, "Fähigkeit nicht gefunden")
|
|
|
|
return r2d(skill)
|
|
|
|
|
|
# ── Create Skill ──────────────────────────────────────────────────────
|
|
@router.post("/skills")
|
|
def create_skill(data: dict, session=Depends(require_auth)):
|
|
"""Create new skill (admin only)."""
|
|
role = session.get('role')
|
|
if role not in ['admin', 'superadmin']:
|
|
raise HTTPException(403, "Nur Admins dürfen Fähigkeiten erstellen")
|
|
|
|
name = data.get('name')
|
|
if not name:
|
|
raise HTTPException(400, "Name ist Pflichtfeld")
|
|
|
|
cat_id = data.get("category_id")
|
|
if cat_id is not None:
|
|
cat_id = int(cat_id)
|
|
main_id = data.get("main_category_id")
|
|
if main_id is not None:
|
|
main_id = int(main_id)
|
|
|
|
focus = data.get("focus_areas")
|
|
if isinstance(focus, (list, dict)):
|
|
focus = json.dumps(focus)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
if cat_id and main_id is None:
|
|
main_id = _main_id_for_category(cur, cat_id)
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO skills (
|
|
name, category, description, importance, keywords, status,
|
|
category_id, main_category_id, focus_areas, sort_order,
|
|
karate_relevance, relevance_level
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
name,
|
|
data.get("category"),
|
|
data.get("description"),
|
|
data.get("importance"),
|
|
_jsonb_param(data.get("keywords")),
|
|
data.get("status", "active"),
|
|
cat_id,
|
|
main_id,
|
|
focus if focus is not None else "[]",
|
|
data.get("sort_order"),
|
|
data.get("karate_relevance"),
|
|
data.get("relevance_level"),
|
|
),
|
|
)
|
|
|
|
skill_id = cur.fetchone()["id"]
|
|
|
|
return get_skill(skill_id, session)
|
|
|
|
|
|
# ── Update Skill ──────────────────────────────────────────────────────
|
|
@router.put("/skills/{skill_id}")
|
|
def update_skill(skill_id: int, data: dict, session=Depends(require_auth)):
|
|
"""Update skill (admin only)."""
|
|
role = session.get('role')
|
|
if role not in ['admin', 'superadmin']:
|
|
raise HTTPException(403, "Nur Admins dürfen Fähigkeiten bearbeiten")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check existence
|
|
cur.execute("SELECT id FROM skills WHERE id = %s", (skill_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Fähigkeit nicht gefunden")
|
|
|
|
sets: list = []
|
|
vals: list = []
|
|
for key in (
|
|
"name",
|
|
"category",
|
|
"description",
|
|
"importance",
|
|
"status",
|
|
"category_id",
|
|
"main_category_id",
|
|
"sort_order",
|
|
"karate_relevance",
|
|
"relevance_level",
|
|
):
|
|
if key in data:
|
|
sets.append(f"{key} = %s")
|
|
vals.append(data[key])
|
|
if "keywords" in data:
|
|
sets.append("keywords = %s")
|
|
vals.append(_jsonb_param(data["keywords"]))
|
|
if "focus_areas" in data:
|
|
fa = data["focus_areas"]
|
|
if isinstance(fa, (list, dict)):
|
|
fa = json.dumps(fa)
|
|
sets.append("focus_areas = %s::jsonb")
|
|
vals.append(fa if fa is not None else "[]")
|
|
|
|
if sets:
|
|
sets.append("updated_at = NOW()")
|
|
vals.append(skill_id)
|
|
cur.execute(
|
|
f"UPDATE skills SET {', '.join(sets)} WHERE id = %s",
|
|
tuple(vals),
|
|
)
|
|
|
|
if "category_id" in data and "main_category_id" not in data:
|
|
mid = _main_id_for_category(cur, data.get("category_id"))
|
|
if mid is not None:
|
|
cur.execute(
|
|
"UPDATE skills SET main_category_id = %s WHERE id = %s",
|
|
(mid, skill_id),
|
|
)
|
|
|
|
return get_skill(skill_id, session)
|
|
|
|
|
|
# ── Delete Skill ──────────────────────────────────────────────────────
|
|
@router.delete("/skills/{skill_id}")
|
|
def delete_skill(skill_id: int, session=Depends(require_auth)):
|
|
"""Delete skill (superadmin only)."""
|
|
role = session.get('role')
|
|
if role != 'superadmin':
|
|
raise HTTPException(403, "Nur Superadmins dürfen Fähigkeiten löschen")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check existence
|
|
cur.execute("SELECT id FROM skills WHERE id = %s", (skill_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Fähigkeit nicht gefunden")
|
|
|
|
# Delete
|
|
cur.execute("DELETE FROM skills WHERE id = %s", (skill_id,))
|
|
|
|
return {"ok": True}
|
|
|
|
|
|
# ── List Training Methods ─────────────────────────────────────────────
|
|
@router.get("/methods")
|
|
def list_methods(
|
|
category: Optional[str] = Query(default=None),
|
|
status: Optional[str] = Query(default=None),
|
|
session=Depends(require_auth)
|
|
):
|
|
"""
|
|
List all training methods (public for authenticated users).
|
|
|
|
Filters:
|
|
- category: kondition, didaktik, koordination, kraft, etc.
|
|
- status: active, inactive
|
|
"""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
query = "SELECT * FROM training_methods"
|
|
params = []
|
|
where = []
|
|
|
|
if category:
|
|
where.append("category = %s")
|
|
params.append(category)
|
|
|
|
if status:
|
|
where.append("status = %s")
|
|
params.append(status)
|
|
else:
|
|
# Default: only active methods
|
|
where.append("status = 'active'")
|
|
|
|
if where:
|
|
query += " WHERE " + " AND ".join(where)
|
|
|
|
query += " ORDER BY name"
|
|
|
|
cur.execute(query, params)
|
|
rows = cur.fetchall()
|
|
return [r2d(r) for r in rows]
|
|
|
|
|
|
# ── Get Training Method ───────────────────────────────────────────────
|
|
@router.get("/methods/{method_id}")
|
|
def get_method(method_id: int, session=Depends(require_auth)):
|
|
"""Get training method by ID."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
cur.execute("SELECT * FROM training_methods WHERE id = %s", (method_id,))
|
|
method = cur.fetchone()
|
|
|
|
if not method:
|
|
raise HTTPException(404, "Trainingsmethode nicht gefunden")
|
|
|
|
return r2d(method)
|
|
|
|
|
|
# ── Create Training Method ────────────────────────────────────────────
|
|
@router.post("/methods")
|
|
def create_method(data: dict, session=Depends(require_auth)):
|
|
"""Create new training method (admin only)."""
|
|
role = session.get('role')
|
|
if role not in ['admin', 'superadmin']:
|
|
raise HTTPException(403, "Nur Admins dürfen Trainingsmethoden erstellen")
|
|
|
|
name = data.get('name')
|
|
if not name:
|
|
raise HTTPException(400, "Name ist Pflichtfeld")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
cur.execute("""
|
|
INSERT INTO training_methods (
|
|
name, abbreviation, category, description,
|
|
typical_duration, typical_group_size,
|
|
related_skills, keywords, status
|
|
) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""", (
|
|
name,
|
|
data.get('abbreviation'),
|
|
data.get('category'),
|
|
data.get('description'),
|
|
data.get('typical_duration'),
|
|
data.get('typical_group_size'),
|
|
data.get('related_skills'),
|
|
data.get('keywords'),
|
|
data.get('status', 'active')
|
|
))
|
|
|
|
method_id = cur.fetchone()['id']
|
|
conn.commit()
|
|
|
|
return get_method(method_id, session)
|
|
|
|
|
|
# ── Update Training Method ────────────────────────────────────────────
|
|
@router.put("/methods/{method_id}")
|
|
def update_method(method_id: int, data: dict, session=Depends(require_auth)):
|
|
"""Update training method (admin only)."""
|
|
role = session.get('role')
|
|
if role not in ['admin', 'superadmin']:
|
|
raise HTTPException(403, "Nur Admins dürfen Trainingsmethoden bearbeiten")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check existence
|
|
cur.execute("SELECT id FROM training_methods WHERE id = %s", (method_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Trainingsmethode nicht gefunden")
|
|
|
|
# Update
|
|
cur.execute("""
|
|
UPDATE training_methods SET
|
|
name = %s,
|
|
abbreviation = %s,
|
|
category = %s,
|
|
description = %s,
|
|
typical_duration = %s,
|
|
typical_group_size = %s,
|
|
related_skills = %s,
|
|
keywords = %s,
|
|
status = %s,
|
|
updated_at = NOW()
|
|
WHERE id = %s
|
|
""", (
|
|
data.get('name'),
|
|
data.get('abbreviation'),
|
|
data.get('category'),
|
|
data.get('description'),
|
|
data.get('typical_duration'),
|
|
data.get('typical_group_size'),
|
|
data.get('related_skills'),
|
|
data.get('keywords'),
|
|
data.get('status'),
|
|
method_id
|
|
))
|
|
|
|
conn.commit()
|
|
|
|
return get_method(method_id, session)
|
|
|
|
|
|
# ── Delete Training Method ────────────────────────────────────────────
|
|
@router.delete("/methods/{method_id}")
|
|
def delete_method(method_id: int, session=Depends(require_auth)):
|
|
"""Delete training method (superadmin only)."""
|
|
role = session.get('role')
|
|
if role != 'superadmin':
|
|
raise HTTPException(403, "Nur Superadmins dürfen Trainingsmethoden löschen")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
|
|
# Check existence
|
|
cur.execute("SELECT id FROM training_methods WHERE id = %s", (method_id,))
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Trainingsmethode nicht gefunden")
|
|
|
|
# Delete
|
|
cur.execute("DELETE FROM training_methods WHERE id = %s", (method_id,))
|
|
conn.commit()
|
|
|
|
return {"ok": True}
|