feat(dashboard): doctor dashboard with red zones, patient card, live alerts and patient connect
CI / ruff · pytest (pull_request) Successful in 2m1s

- Patient list sorted red -> yellow -> grey -> green with KPI row, trend arrows,
  7-day adherence bars, discontinuation-risk badge with factor tooltip.
- Patient card: scales with severity bands from config/scales (unscheduled marked),
  mood/sleep/alcohol small multiples, dose heat-strip, alerts feed with ack,
  safety plan, AI summary with disclaimer.
- Live alerts: st.fragment(run_every=2.5s) polling, toast + pinned acute banner on
  every page; full rerun on new alert only on doctor pages.
- Connect form -> POST /patients/connect -> code, QR, deep link.
- Thin httpx ApiClient over the frozen schemas with graceful "API недоступен".
- app.py discovers dashboard/pages/*.py automatically (st.navigation); demo pages
  of #17 plug in by adding a file.

Refs #16

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-04 12:51:07 +03:00
co-authored by Claude Opus 5.5
parent 178cab5b1b
commit 2582c73d12
15 changed files with 2170 additions and 13 deletions
+132
View File
@@ -0,0 +1,132 @@
"""Thin HTTP client for the dashboard. The frozen API schemas (`app/api/schemas.py`) are the only contract.
The dashboard never touches the databases: everything goes over HTTP to `API_BASE_URL`. Short timeouts keep the
UI responsive when the backend is down; every failure is turned into `ApiError` with a Russian message that the
page shows as a banner instead of a traceback.
"""
from __future__ import annotations
import logging
import os
from typing import Any, TypeVar
import httpx
from pydantic import BaseModel, TypeAdapter, ValidationError
from app.api.schemas import (
AckIn,
AlertOut,
ClockOut,
ConnectPatientIn,
ConnectPatientOut,
PatientCard,
PatientListItem,
)
log = logging.getLogger(__name__)
DEFAULT_BASE_URL = "http://localhost:8000"
DEFAULT_DOCTOR_ID = "doc-smirnova"
API_PREFIX = "/api/v1"
T = TypeVar("T")
class ApiError(Exception):
"""Any API failure, with a message safe to show to the doctor (no PII, no stack traces)."""
def __init__(self, message: str, status: int | None = None) -> None:
super().__init__(message)
self.message = message
self.status = status
class ApiUnavailable(ApiError):
"""Backend not reachable (connection refused, DNS, timeout)."""
class ApiClient:
def __init__(
self,
base_url: str | None = None,
doctor_id: str | None = None,
*,
timeout: float = 3.0,
transport: httpx.BaseTransport | None = None,
) -> None:
base = (base_url or os.environ.get("API_BASE_URL") or DEFAULT_BASE_URL).rstrip("/")
if not base.endswith(API_PREFIX):
base += API_PREFIX
self.base_url = base
self.doctor_id = doctor_id or os.environ.get("DOCTOR_ID") or DEFAULT_DOCTOR_ID
self._http = httpx.Client(
base_url=base, timeout=httpx.Timeout(timeout, connect=min(timeout, 1.5)), transport=transport
)
# ------------------------------------------------------------------ doctor endpoints
def patients(self) -> list[PatientListItem]:
return self._get(f"/doctors/{self.doctor_id}/patients", list[PatientListItem])
def patient_card(self, token: str) -> PatientCard:
return self._get(f"/patients/{token}", PatientCard, params={"doctor_id": self.doctor_id})
def alerts(self, since_id: int = -1, *, open_only: bool = False) -> list[AlertOut]:
params = {"since_id": since_id, "open_only": str(open_only).lower()}
return self._get(f"/doctors/{self.doctor_id}/alerts", list[AlertOut], params=params)
def ack(self, alert_id: int) -> AlertOut:
return self._request("POST", f"/alerts/{alert_id}/ack", AlertOut, json=AckIn(doctor_id=self.doctor_id))
def connect(self, body: ConnectPatientIn) -> ConnectPatientOut:
return self._request("POST", "/patients/connect", ConnectPatientOut, json=body)
def clock(self) -> ClockOut | None:
"""Demo clock (ADR-005); optional decoration, so failures return None instead of raising."""
try:
return self._get("/demo/clock", ClockOut)
except ApiError:
return None
# ------------------------------------------------------------------ plumbing
def _get(self, path: str, model: type[T], params: dict[str, Any] | None = None) -> T:
return self._request("GET", path, model, params=params)
def _request(
self,
method: str,
path: str,
model: type[T],
*,
params: dict[str, Any] | None = None,
json: BaseModel | None = None,
) -> T:
payload = json.model_dump(mode="json") if json is not None else None
try:
resp = self._http.request(method, path, params=params, json=payload)
except (httpx.ConnectError, httpx.TimeoutException) as exc:
log.warning("api unavailable: %s %s (%s)", method, _route(path), type(exc).__name__)
raise ApiUnavailable("API недоступен — проверьте, что сервис запущен") from exc
except httpx.HTTPError as exc:
log.warning("api transport error: %s %s (%s)", method, _route(path), type(exc).__name__)
raise ApiUnavailable("API недоступен — ошибка соединения") from exc
if resp.status_code == 404:
raise ApiError("Не найдено", 404)
if resp.status_code >= 400:
log.warning("api error: %s %s -> %s", method, _route(path), resp.status_code)
raise ApiError(f"Ошибка API ({resp.status_code})", resp.status_code)
try:
return TypeAdapter(model).validate_python(resp.json())
except (ValueError, ValidationError) as exc:
log.warning("api bad payload: %s %s", method, _route(path))
raise ApiError("API вернул неожиданный ответ") from exc
def close(self) -> None:
self._http.close()
def _route(path: str) -> str:
"""Path without identifiers, so logs carry no patient tokens."""
parts = path.strip("/").split("/")
return "/" + "/".join(p if not any(ch.isdigit() for ch in p) and not p.startswith("tok") else "*" for p in parts)