Lars
3e916c082c
- Introduced a new API endpoint for fetching a summary of profile reference values, providing the latest and previous entries for each reference type. - Updated ProfileReferenceValuesPage to display summary tiles with trend indicators for better user insights. - Enhanced CSS for responsive layout of reference value tiles, improving the overall user experience on different screen sizes. - Implemented trend calculation logic to visually represent changes between the latest and previous reference values.
446 lines
15 KiB
Python
446 lines
15 KiB
Python
"""
|
|
Persönliche Referenzwerte (profilorientiert)
|
|
|
|
Typkatalog system-seeded; Nutzer pflegt historische Einträge pro aktivem Profil.
|
|
Einheit immer aus dem Typ; Wert je value_data_type validiert.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Header, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from psycopg2.extras import Json
|
|
|
|
from auth import require_auth
|
|
from db import get_db, get_cursor, r2d
|
|
from reference_value_validation import (
|
|
REF_VALUE_CONFIDENCE,
|
|
REF_VALUE_CONFIDENCE_ORDER,
|
|
REF_VALUE_METHODS,
|
|
REF_VALUE_SOURCES,
|
|
validate_meta_confidence,
|
|
validate_meta_method,
|
|
validate_meta_source,
|
|
validate_value_for_data_type,
|
|
resolve_unit_from_type,
|
|
)
|
|
from routers.profiles import get_pid
|
|
|
|
router = APIRouter(prefix="/api", tags=["reference-values"])
|
|
|
|
|
|
def _row_to_api(d: dict[str, Any]) -> dict[str, Any]:
|
|
if not d:
|
|
return d
|
|
out = dict(d)
|
|
ed = out.get("effective_date")
|
|
if ed is not None and hasattr(ed, "isoformat"):
|
|
out["effective_date"] = ed.isoformat()
|
|
ca = out.get("created_at")
|
|
if ca is not None and hasattr(ca, "isoformat"):
|
|
out["created_at"] = ca.isoformat()
|
|
ua = out.get("updated_at")
|
|
if ua is not None and hasattr(ua, "isoformat"):
|
|
out["updated_at"] = ua.isoformat()
|
|
vn = out.get("value_numeric")
|
|
if vn is not None and isinstance(vn, Decimal):
|
|
out["value_numeric"] = float(vn)
|
|
return out
|
|
|
|
|
|
def _get_type_by_key(cur, key: str, require_active: bool = True) -> dict:
|
|
q = (
|
|
"SELECT id, key, label, description, default_unit, active, category, "
|
|
"value_data_type, validation_rules, metadata "
|
|
"FROM reference_value_types WHERE key = %s "
|
|
)
|
|
if require_active:
|
|
q += "AND active = TRUE "
|
|
cur.execute(q, (key,))
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Referenztyp nicht gefunden")
|
|
return r2d(row)
|
|
|
|
|
|
class ProfileReferenceValueCreate(BaseModel):
|
|
reference_value_type_key: str = Field(..., min_length=1, max_length=64)
|
|
effective_date: str
|
|
value_numeric: Optional[float] = None
|
|
value_text: Optional[str] = None
|
|
source: str = Field(..., min_length=1)
|
|
method: str = Field(..., min_length=1)
|
|
confidence: str = Field(..., min_length=1)
|
|
notes: Optional[str] = None
|
|
extra: Optional[dict] = None
|
|
|
|
|
|
class ProfileReferenceValueUpdate(BaseModel):
|
|
effective_date: Optional[str] = None
|
|
value_numeric: Optional[float] = None
|
|
value_text: Optional[str] = None
|
|
source: Optional[str] = None
|
|
method: Optional[str] = None
|
|
confidence: Optional[str] = None
|
|
notes: Optional[str] = None
|
|
extra: Optional[dict] = None
|
|
|
|
|
|
@router.get("/reference-value-types")
|
|
def list_reference_value_types(session: dict = Depends(require_auth)):
|
|
"""Alle aktiven Referenztyp-Definitionen (dynamische UI inkl. Validierungsmetadaten)."""
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
id, key, label, description, default_unit, sort_order, active,
|
|
category, value_data_type, validation_rules, metadata, created_at
|
|
FROM reference_value_types
|
|
WHERE active = TRUE
|
|
ORDER BY sort_order ASC, id ASC
|
|
"""
|
|
)
|
|
return [_row_to_api(r2d(r)) for r in cur.fetchall()]
|
|
|
|
|
|
@router.get("/reference-value-meta/enums")
|
|
def list_reference_value_meta_enums(session: dict = Depends(require_auth)):
|
|
"""Erlaubte Werte für Quelle, Methode, Vertrauensgrad (Erfassungsdialog)."""
|
|
return {
|
|
"sources": sorted(REF_VALUE_SOURCES),
|
|
"methods": sorted(REF_VALUE_METHODS),
|
|
"confidence_levels": [x for x in REF_VALUE_CONFIDENCE_ORDER if x in REF_VALUE_CONFIDENCE],
|
|
}
|
|
|
|
|
|
@router.get("/profile-reference-values/summary")
|
|
def profile_reference_values_summary(
|
|
x_profile_id: Optional[str] = Header(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""
|
|
Für das aktive Profil: je Referenztyp mit mindestens einem Eintrag der jüngste Wert
|
|
plus der unmittelbar vorherige (gleiche Sortierung wie Liste), für Tendenz-Anzeigen.
|
|
"""
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
WITH ranked AS (
|
|
SELECT
|
|
v.id,
|
|
v.profile_id,
|
|
v.reference_value_type_id,
|
|
v.effective_date,
|
|
v.value_numeric,
|
|
v.value_text,
|
|
v.unit,
|
|
v.source,
|
|
v.confidence,
|
|
v.method,
|
|
v.notes,
|
|
v.extra,
|
|
v.created_at,
|
|
v.updated_at,
|
|
rt.key AS type_key,
|
|
rt.label AS type_label,
|
|
rt.sort_order AS type_sort_order,
|
|
rt.value_data_type,
|
|
ROW_NUMBER() OVER (
|
|
PARTITION BY v.reference_value_type_id
|
|
ORDER BY v.effective_date DESC, v.created_at DESC
|
|
) AS rn
|
|
FROM profile_reference_values v
|
|
JOIN reference_value_types rt ON rt.id = v.reference_value_type_id
|
|
WHERE v.profile_id = %s AND rt.active = TRUE
|
|
)
|
|
SELECT * FROM ranked WHERE rn <= 2
|
|
ORDER BY type_sort_order ASC, type_key ASC, rn ASC
|
|
""",
|
|
(pid,),
|
|
)
|
|
raw_rows = [r2d(r) for r in cur.fetchall()]
|
|
|
|
by_key: dict[str, dict[str, Any]] = {}
|
|
skip_cols = frozenset({"rn", "type_sort_order", "value_data_type"})
|
|
for row in raw_rows:
|
|
rn = int(row.get("rn") or 0)
|
|
key = row["type_key"]
|
|
if key not in by_key:
|
|
by_key[key] = {
|
|
"type_key": key,
|
|
"type_label": row.get("type_label") or key,
|
|
"value_data_type": (row.get("value_data_type") or "decimal").strip().lower(),
|
|
"sort_key": (row.get("type_sort_order") or 0, key),
|
|
"latest": None,
|
|
"previous": None,
|
|
}
|
|
entry = {k: v for k, v in row.items() if k not in skip_cols}
|
|
api_entry = _row_to_api(entry)
|
|
if rn == 1:
|
|
by_key[key]["latest"] = api_entry
|
|
elif rn == 2:
|
|
by_key[key]["previous"] = api_entry
|
|
|
|
tiles = sorted(by_key.values(), key=lambda t: t["sort_key"])
|
|
for t in tiles:
|
|
t.pop("sort_key", None)
|
|
|
|
return {"tiles": tiles}
|
|
|
|
|
|
@router.get("/profile-reference-values")
|
|
def list_profile_reference_values(
|
|
type_key: str = Query(..., description="Schlüssel aus reference_value_types.key"),
|
|
x_profile_id: Optional[str] = Header(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
"""Historische Einträge eines Typs für das aktive Profil (neueste zuerst)."""
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
t = _get_type_by_key(cur, type_key, require_active=True)
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
v.id,
|
|
v.profile_id,
|
|
v.reference_value_type_id,
|
|
v.effective_date,
|
|
v.value_numeric,
|
|
v.value_text,
|
|
v.unit,
|
|
v.source,
|
|
v.confidence,
|
|
v.method,
|
|
v.notes,
|
|
v.extra,
|
|
v.created_at,
|
|
v.updated_at,
|
|
rt.key AS type_key,
|
|
rt.label AS type_label
|
|
FROM profile_reference_values v
|
|
JOIN reference_value_types rt ON rt.id = v.reference_value_type_id
|
|
WHERE v.profile_id = %s AND rt.key = %s
|
|
ORDER BY v.effective_date DESC, v.created_at DESC
|
|
""",
|
|
(pid, t["key"]),
|
|
)
|
|
return [_row_to_api(r2d(r)) for r in cur.fetchall()]
|
|
|
|
|
|
@router.post("/profile-reference-values")
|
|
def create_profile_reference_value(
|
|
body: ProfileReferenceValueCreate,
|
|
x_profile_id: Optional[str] = Header(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
pid = get_pid(x_profile_id)
|
|
try:
|
|
datetime.strptime(body.effective_date, "%Y-%m-%d")
|
|
except ValueError:
|
|
raise HTTPException(400, "Ungültiges Datum. Format: YYYY-MM-DD")
|
|
|
|
src = validate_meta_source(body.source)
|
|
meth = validate_meta_method(body.method)
|
|
conf = validate_meta_confidence(body.confidence)
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
t = _get_type_by_key(cur, body.reference_value_type_key.strip(), require_active=True)
|
|
vdt = (t.get("value_data_type") or "decimal").strip().lower()
|
|
rules = t.get("validation_rules") or {}
|
|
vnum, vtxt = validate_value_for_data_type(vdt, rules, body.value_numeric, body.value_text)
|
|
unit = resolve_unit_from_type(t.get("default_unit"))
|
|
extra = body.extra if body.extra is not None else {}
|
|
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO profile_reference_values (
|
|
profile_id, reference_value_type_id, effective_date,
|
|
value_numeric, value_text, unit, source, confidence, method, notes, extra
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
|
|
RETURNING id
|
|
""",
|
|
(
|
|
pid,
|
|
t["id"],
|
|
body.effective_date,
|
|
vnum,
|
|
vtxt,
|
|
unit,
|
|
src,
|
|
conf,
|
|
meth,
|
|
body.notes,
|
|
Json(extra),
|
|
),
|
|
)
|
|
new_id = cur.fetchone()["id"]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
v.id,
|
|
v.profile_id,
|
|
v.reference_value_type_id,
|
|
v.effective_date,
|
|
v.value_numeric,
|
|
v.value_text,
|
|
v.unit,
|
|
v.source,
|
|
v.confidence,
|
|
v.method,
|
|
v.notes,
|
|
v.extra,
|
|
v.created_at,
|
|
v.updated_at,
|
|
rt.key AS type_key,
|
|
rt.label AS type_label
|
|
FROM profile_reference_values v
|
|
JOIN reference_value_types rt ON rt.id = v.reference_value_type_id
|
|
WHERE v.id = %s AND v.profile_id = %s
|
|
""",
|
|
(new_id, pid),
|
|
)
|
|
return _row_to_api(r2d(cur.fetchone()))
|
|
|
|
|
|
@router.put("/profile-reference-values/{entry_id}")
|
|
def update_profile_reference_value(
|
|
entry_id: int,
|
|
body: ProfileReferenceValueUpdate,
|
|
x_profile_id: Optional[str] = Header(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
pid = get_pid(x_profile_id)
|
|
patch = body.model_dump(exclude_unset=True)
|
|
if not patch:
|
|
raise HTTPException(400, "Keine Felder zum Aktualisieren")
|
|
|
|
if patch.get("effective_date"):
|
|
try:
|
|
datetime.strptime(patch["effective_date"], "%Y-%m-%d")
|
|
except ValueError:
|
|
raise HTTPException(400, "Ungültiges Datum. Format: YYYY-MM-DD")
|
|
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"""
|
|
SELECT v.*, rt.default_unit, rt.value_data_type, rt.validation_rules
|
|
FROM profile_reference_values v
|
|
JOIN reference_value_types rt ON rt.id = v.reference_value_type_id
|
|
WHERE v.id = %s AND v.profile_id = %s
|
|
""",
|
|
(entry_id, pid),
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
raise HTTPException(404, "Eintrag nicht gefunden")
|
|
cur_row = r2d(row)
|
|
|
|
vdt = (cur_row.get("value_data_type") or "decimal").strip().lower()
|
|
rules = cur_row.get("validation_rules") or {}
|
|
|
|
new_ed = patch.get("effective_date", cur_row["effective_date"])
|
|
if hasattr(new_ed, "isoformat"):
|
|
new_ed = new_ed.isoformat()
|
|
|
|
vn = patch["value_numeric"] if "value_numeric" in patch else cur_row.get("value_numeric")
|
|
vt_raw = patch["value_text"] if "value_text" in patch else cur_row.get("value_text")
|
|
if vn is not None and isinstance(vn, Decimal):
|
|
vn = float(vn)
|
|
|
|
vnum, vtxt = validate_value_for_data_type(vdt, rules, vn, vt_raw)
|
|
|
|
unit = resolve_unit_from_type(cur_row.get("default_unit"))
|
|
|
|
if "source" in patch:
|
|
src = validate_meta_source(patch["source"])
|
|
else:
|
|
src = validate_meta_source(cur_row.get("source"))
|
|
if "method" in patch:
|
|
meth = validate_meta_method(patch["method"])
|
|
else:
|
|
meth = validate_meta_method(cur_row.get("method"))
|
|
if "confidence" in patch:
|
|
conf = validate_meta_confidence(patch["confidence"])
|
|
else:
|
|
conf = validate_meta_confidence(cur_row.get("confidence"))
|
|
|
|
updates: dict[str, Any] = {
|
|
"effective_date": new_ed,
|
|
"value_numeric": vnum,
|
|
"value_text": vtxt,
|
|
"unit": unit,
|
|
"source": src,
|
|
"method": meth,
|
|
"confidence": conf,
|
|
}
|
|
if "notes" in patch:
|
|
updates["notes"] = patch["notes"]
|
|
if "extra" in patch:
|
|
updates["extra"] = Json(patch["extra"] if patch["extra"] is not None else {})
|
|
|
|
set_parts = [f"{k} = %s" for k in updates]
|
|
vals = list(updates.values()) + [entry_id, pid]
|
|
cur.execute(
|
|
f"""
|
|
UPDATE profile_reference_values SET {", ".join(set_parts)}, updated_at = NOW()
|
|
WHERE id = %s AND profile_id = %s
|
|
""",
|
|
tuple(vals),
|
|
)
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
v.id,
|
|
v.profile_id,
|
|
v.reference_value_type_id,
|
|
v.effective_date,
|
|
v.value_numeric,
|
|
v.value_text,
|
|
v.unit,
|
|
v.source,
|
|
v.confidence,
|
|
v.method,
|
|
v.notes,
|
|
v.extra,
|
|
v.created_at,
|
|
v.updated_at,
|
|
rt.key AS type_key,
|
|
rt.label AS type_label
|
|
FROM profile_reference_values v
|
|
JOIN reference_value_types rt ON rt.id = v.reference_value_type_id
|
|
WHERE v.id = %s AND v.profile_id = %s
|
|
""",
|
|
(entry_id, pid),
|
|
)
|
|
return _row_to_api(r2d(cur.fetchone()))
|
|
|
|
|
|
@router.delete("/profile-reference-values/{entry_id}")
|
|
def delete_profile_reference_value(
|
|
entry_id: int,
|
|
x_profile_id: Optional[str] = Header(default=None),
|
|
session: dict = Depends(require_auth),
|
|
):
|
|
pid = get_pid(x_profile_id)
|
|
with get_db() as conn:
|
|
cur = get_cursor(conn)
|
|
cur.execute(
|
|
"DELETE FROM profile_reference_values WHERE id = %s AND profile_id = %s RETURNING id",
|
|
(entry_id, pid),
|
|
)
|
|
if not cur.fetchone():
|
|
raise HTTPException(404, "Eintrag nicht gefunden")
|
|
return {"ok": True}
|