78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
|
"""
|
||
|
|
Persistenter System-Standard für die Produkt-Übersicht (Dashboard).
|
||
|
|
|
||
|
|
Key in system_config: dashboard_product_default — gültiges DashboardLayoutPayload (JSON).
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from psycopg2.extras import Json
|
||
|
|
|
||
|
|
from dashboard_layout_schema import DashboardLayoutPayload, product_default_layout_dict
|
||
|
|
from db import get_cursor
|
||
|
|
|
||
|
|
SYSTEM_CONFIG_KEY_DASHBOARD_PRODUCT_DEFAULT = "dashboard_product_default"
|
||
|
|
|
||
|
|
|
||
|
|
def get_stored_product_default_validated(conn) -> dict[str, Any] | None:
|
||
|
|
"""Gültiges Layout aus DB oder None (fehlt/ungültig)."""
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute(
|
||
|
|
"SELECT value FROM system_config WHERE key = %s",
|
||
|
|
(SYSTEM_CONFIG_KEY_DASHBOARD_PRODUCT_DEFAULT,),
|
||
|
|
)
|
||
|
|
row = cur.fetchone()
|
||
|
|
if not row or row.get("value") is None:
|
||
|
|
return None
|
||
|
|
raw = row["value"]
|
||
|
|
if isinstance(raw, str):
|
||
|
|
import json
|
||
|
|
|
||
|
|
try:
|
||
|
|
raw = json.loads(raw)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
return None
|
||
|
|
if not isinstance(raw, dict):
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
payload = DashboardLayoutPayload.model_validate(
|
||
|
|
{"version": raw.get("version", 1), "widgets": raw.get("widgets", [])}
|
||
|
|
)
|
||
|
|
return payload.to_stored_dict()
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def get_product_default_base_dict(conn) -> dict[str, Any]:
|
||
|
|
"""Basis-Layout (ohne Entitlements): DB-Override oder Code-Standard."""
|
||
|
|
stored = get_stored_product_default_validated(conn)
|
||
|
|
if stored is not None:
|
||
|
|
return stored
|
||
|
|
return product_default_layout_dict()
|
||
|
|
|
||
|
|
|
||
|
|
def upsert_product_default_base(conn, layout: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
payload = DashboardLayoutPayload.model_validate(layout)
|
||
|
|
stored = payload.to_stored_dict()
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute(
|
||
|
|
"""
|
||
|
|
INSERT INTO system_config (key, value, updated_at)
|
||
|
|
VALUES (%s, %s, CURRENT_TIMESTAMP)
|
||
|
|
ON CONFLICT (key) DO UPDATE SET
|
||
|
|
value = EXCLUDED.value,
|
||
|
|
updated_at = CURRENT_TIMESTAMP
|
||
|
|
""",
|
||
|
|
(SYSTEM_CONFIG_KEY_DASHBOARD_PRODUCT_DEFAULT, Json(stored)),
|
||
|
|
)
|
||
|
|
return stored
|
||
|
|
|
||
|
|
|
||
|
|
def delete_product_default_override(conn) -> None:
|
||
|
|
cur = get_cursor(conn)
|
||
|
|
cur.execute(
|
||
|
|
"DELETE FROM system_config WHERE key = %s",
|
||
|
|
(SYSTEM_CONFIG_KEY_DASHBOARD_PRODUCT_DEFAULT,),
|
||
|
|
)
|