57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
from __future__ import annotations
|
||
|
||
import logging
|
||
from collections.abc import Coroutine
|
||
from dataclasses import asdict, is_dataclass
|
||
from decimal import Decimal
|
||
from typing import Any, TypeVar
|
||
|
||
import streamlit as st
|
||
|
||
from app.db.base import run_db
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
T = TypeVar("T")
|
||
|
||
|
||
def current_user() -> str:
|
||
"""email из профиля Keycloak (AuthService гарантирует, что он есть)."""
|
||
profile = st.session_state.get("user_profile", {})
|
||
if isinstance(profile, dict) and profile.get("email"):
|
||
return str(profile["email"])
|
||
raise RuntimeError("Пользователь не авторизован.")
|
||
|
||
|
||
def call_db(coro: Coroutine[Any, Any, T], error_message: str) -> T | None:
|
||
"""run_db() для экранов: непредвиденная ошибка -> st.error, а не трейсбек.
|
||
|
||
Ожидаемые ошибки сервисов (с понятным текстом) ловятся на месте вызова.
|
||
"""
|
||
try:
|
||
return run_db(coro)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.exception(error_message)
|
||
st.error(f"{error_message}: {exc}")
|
||
return None
|
||
|
||
|
||
def as_table(rows: list[Any], columns: dict[str, str]) -> list[dict[str, Any]]:
|
||
"""Список dataclass -> строки для st.dataframe с русскими заголовками.
|
||
|
||
Decimal выводим строкой: так видно ровно то, что лежит в БД, без float.
|
||
"""
|
||
table = []
|
||
for row in rows:
|
||
data = asdict(row) if is_dataclass(row) else dict(row)
|
||
table.append(
|
||
{title: _display(data.get(name)) for name, title in columns.items()}
|
||
)
|
||
return table
|
||
|
||
|
||
def _display(value: Any) -> Any:
|
||
if isinstance(value, Decimal):
|
||
return format(value.normalize(), "f")
|
||
return value
|