Lars
24daeeb83c
- Added new API endpoints for listing and updating widget-feature assignments, allowing for custom feature requirements. - Introduced a new admin page for managing widget-feature assignments, enhancing the admin interface. - Updated navigation to include a link to the new widget-feature assignments page. - Refactored widget access logic to support AND-based feature requirements for widgets. - Bumped app_dashboard version to 1.11.0 to reflect these changes and improvements.
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""
|
|
DB-Override für Dashboard-Widget → Feature(s): AND-Semantik.
|
|
|
|
Ohne Zeile in dashboard_widget_requirement_custom → Fallback auf widget_catalog.requires_feature.
|
|
Mit Marker-Zeile → nur widget_feature_requirements (0..n, leer = kein Feature erforderlich).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from db import get_cursor, get_db
|
|
from widget_catalog import WIDGET_CATALOG
|
|
|
|
_WIDGET_ENTRY_BY_ID: dict[str, dict[str, Any]] = {e["id"]: e for e in WIDGET_CATALOG}
|
|
|
|
|
|
def _catalog_feature_ids_for_widget(widget_id: str) -> list[str]:
|
|
entry = _WIDGET_ENTRY_BY_ID.get(widget_id)
|
|
if not entry:
|
|
return []
|
|
fid = entry.get("requires_feature")
|
|
return [fid] if fid else []
|
|
|
|
|
|
def get_widget_required_feature_ids(widget_id: str, conn: Any | None) -> list[str]:
|
|
"""Liste der Feature-IDs, die alle erlaubt sein müssen (AND). Leer = ohne Feature-Gate."""
|
|
|
|
def _query(c: Any) -> list[str]:
|
|
cur = get_cursor(c)
|
|
cur.execute(
|
|
"SELECT 1 FROM dashboard_widget_requirement_custom WHERE widget_id = %s",
|
|
(widget_id,),
|
|
)
|
|
if cur.fetchone() is None:
|
|
return _catalog_feature_ids_for_widget(widget_id)
|
|
cur.execute(
|
|
"""
|
|
SELECT feature_id FROM widget_feature_requirements
|
|
WHERE widget_id = %s
|
|
ORDER BY feature_id
|
|
""",
|
|
(widget_id,),
|
|
)
|
|
return [row["feature_id"] for row in cur.fetchall()]
|
|
|
|
if conn is not None:
|
|
return _query(conn)
|
|
with get_db() as c:
|
|
return _query(c)
|
|
|
|
|
|
def fetch_assignments_bundle(conn: Any) -> tuple[set[str], dict[str, list[str]]]:
|
|
"""
|
|
Alle Custom-Marker und Junction-Zeilen für Admin-GET (ein Roundtrip).
|
|
Returns (custom_widget_ids, feature_ids_by_widget).
|
|
"""
|
|
cur = get_cursor(conn)
|
|
cur.execute("SELECT widget_id FROM dashboard_widget_requirement_custom")
|
|
custom_ids = {row["widget_id"] for row in cur.fetchall()}
|
|
cur.execute(
|
|
"""
|
|
SELECT widget_id, feature_id FROM widget_feature_requirements
|
|
ORDER BY widget_id, feature_id
|
|
"""
|
|
)
|
|
by_w: dict[str, list[str]] = {}
|
|
for row in cur.fetchall():
|
|
by_w.setdefault(row["widget_id"], []).append(row["feature_id"])
|
|
return custom_ids, by_w
|
|
|
|
|
|
def set_custom_requirements(conn: Any, widget_id: str, feature_ids: list[str]) -> None:
|
|
"""Custom aktivieren und Anforderungen ersetzen (dedupliziert)."""
|
|
seen: set[str] = set()
|
|
unique: list[str] = []
|
|
for fid in feature_ids:
|
|
if fid and fid not in seen:
|
|
seen.add(fid)
|
|
unique.append(fid)
|
|
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO dashboard_widget_requirement_custom (widget_id)
|
|
VALUES (%s)
|
|
ON CONFLICT (widget_id) DO UPDATE SET updated_at = CURRENT_TIMESTAMP
|
|
""",
|
|
(widget_id,),
|
|
)
|
|
cur.execute("DELETE FROM widget_feature_requirements WHERE widget_id = %s", (widget_id,))
|
|
for fid in unique:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO widget_feature_requirements (widget_id, feature_id)
|
|
VALUES (%s, %s)
|
|
""",
|
|
(widget_id, fid),
|
|
)
|
|
|
|
|
|
def clear_custom_requirements(conn: Any, widget_id: str) -> None:
|
|
"""Zurück auf Katalog-Fallback."""
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM dashboard_widget_requirement_custom WHERE widget_id = %s",
|
|
(widget_id,),
|
|
)
|