Exceptions: further reworking how GUI and server handle exceptions, including developer dialog that allows you to search error codes in the GUI.

This commit is contained in:
2026-02-26 17:17:32 +01:00
parent 2dd3f0d9df
commit dee0405e2e
5 changed files with 623 additions and 45 deletions
+80 -1
View File
@@ -27,8 +27,87 @@ class AuthErrorCode(StrEnum):
NOT_IN_ACTIVE_PGROUP = "NOT_IN_ACTIVE_PGROUP"
class DAQErrorCode(StrEnum):
BEAMLINE_BUSY = "BEAMLINE_BUSY"
_ERROR_CODE_HELP: dict[str, str] = {
# Auth/JWT
AuthErrorCode.AUTHENTICATION_ERROR: (
"Generic authentication problem. Usually means the request lacked valid credentials "
"(expired/invalid token, missing Authorization header, etc.)."
),
AuthErrorCode.AUTHENTICATION_FAILED: (
"Authentication failed during login/token creation. Typically incorrect credentials "
"or an inability to validate the user."
),
AuthErrorCode.INVALID_TOKEN: (
"The provided token could not be decoded/validated (bad signature, expired, malformed). "
"Re-authenticate to obtain a new token."
),
AuthErrorCode.SESSION_ALREADY_ACTIVE: (
"A different session currently owns control. Use “force current session” (if allowed) "
"or wait for the active session to expire/end."
),
# Authorization
AuthErrorCode.FORBIDDEN: (
"Generic permissions failure. The user is authenticated but not allowed to perform this action."
),
AuthErrorCode.NOT_STAFF: (
"This action requires staff privileges. Log in with a staff account or ask staff to perform it."
),
AuthErrorCode.NOT_IN_ACTIVE_PGROUP: (
"You are not a member of the currently active p-group. Change p-group or use an account "
"that belongs to the active group."
),
# Generic
AuthErrorCode.HTTP_ERROR: (
"Generic HTTP error wrapper. The server returned an HTTPException that wasnt mapped to a more specific code."
),
AuthErrorCode.INTERNAL_SERVER_ERROR: (
"Unhandled server error. Check server logs for a stack trace and context."
),
DAQErrorCode.BEAMLINE_BUSY: (
"Beamline state is set to Busy by prior action. If this state persists an additional error may have occurred, "
"preventing the state from being released, this should timeout within 10 minutes."
"If this occurs please seek assistance from your local contact."
)
}
def error_code_help(code: str) -> str | None:
"""
Return a human help message for a code string, if known.
Accepts either enum value strings or raw strings.
"""
if not code:
return None
return _ERROR_CODE_HELP.get(str(code))
def export_error_code_help() -> dict[str, str]:
"""
Export help text as {"CODE": "help text", ...}
"""
return {str(k): str(v) for k, v in _ERROR_CODE_HELP.items()}
def export_error_codes_grouped() -> dict[str, dict[str, str]]:
"""
Export codes grouped by enum class name:
{
"AuthErrorCode": {"INVALID_TOKEN": "INVALID_TOKEN", ...},
"DAQErrorCode": {"BEAMLINE_BUSY": "BEAMLINE_BUSY", ...}
}
"""
enums: tuple[type[StrEnum], ...] = (AuthErrorCode, DAQErrorCode)
return {e.__name__: {c.name: str(c.value) for c in e} for e in enums}
def export_error_codes() -> dict[str, str]:
"""
Convenience for clients/tests/docs: {"INVALID_TOKEN": "INVALID_TOKEN", ...}
Backwards-compatible, flat export used by older clients/tests/docs:
{"INVALID_TOKEN": "INVALID_TOKEN", ...}
NOTE: This intentionally exports only AuthErrorCode to avoid breaking
existing consumers that assume a flat map and/or specific keys.
"""
return {c.name: str(c.value) for c in AuthErrorCode}
+39 -35
View File
@@ -82,10 +82,23 @@ class BeamlineConfig:
@property
def active_session(self) -> int | None:
tmp = self.__client.get(f"{self.__bl}:active_session")
if tmp is None:
"""
Read active_session from Redis and convert to int.
Returns:
int if present and valid, otherwise None.
"""
raw = self.__client.get(f"{self.__bl}:active_session")
if raw is None:
return None
try:
return int(raw)
except (TypeError, ValueError):
logger.warning(
"Invalid active_session value in redis; treating as missing",
extra={"beamline": self.__bl, "raw": raw},
)
return None
return int(tmp)
def session_status(self, session: int) -> SessionStatus:
return SessionStatus(session=self.session_state(session),
@@ -102,12 +115,12 @@ class BeamlineConfig:
def try_set_active_session(self, session: int, expiry_sec: int) -> None:
with redis_lock.Lock(
self.__client, f"{self.__bl}:active_session_lock", expire=10
self.__client, f"{self.__bl}:active_session_lock", expire=10
):
tmp = self.__client.get(f"{self.__bl}:active_session")
if tmp is None:
active = self.active_session
if active is None:
self.__client.set(f"{self.__bl}:active_session", session)
elif int(tmp) != session:
elif active != session:
raise Exception(
"There is already active session with different id. Try again later."
)
@@ -116,37 +129,27 @@ class BeamlineConfig:
#TODO finish setting this up!
def try_extend_active_session(self, session: int, expiry_sec: int) -> None:
with redis_lock.Lock(
self.__client, f"{self.__bl}:active_session_lock", expire=10
self.__client, f"{self.__bl}:active_session_lock", expire=10
):
tmp = self.__client.get(f"{self.__bl}:active_session")
if tmp == session:
self.__client.expire(f"{self.__bl}:active_session", expiry_sec, gt=True)
elif int(tmp) != session:
raise Exception(
"There is already active session with different id. Try again later."
)
else:
active = self.active_session
if active is None:
raise Exception(
"There is no active session with given id. Try again later."
)
if active == session:
self.__client.expire(f"{self.__bl}:active_session", expiry_sec, gt=True)
else:
raise Exception(
"There is already active session with different id. Try again later."
)
def end_active_session(self, session: int) -> None:
with redis_lock.Lock(
self.__client, f"{self.__bl}:active_session_lock", expire=10
):
raw = self.__client.get(f"{self.__bl}:active_session")
if raw is None:
active = self.active_session
if active is None:
return
try:
active = int(raw)
except (TypeError, ValueError):
logger.warning(
"Invalid active_session value in redis; ignoring",
extra={"beamline": self.__bl, "raw": raw},
)
return
if active == session:
self.__client.delete(f"{self.__bl}:active_session")
@@ -564,12 +567,6 @@ class BeamlineConfig:
else:
self.__client.set(f"{self.__bl}:last_best_b_factor", last_best_b_factor)
@crystal_size.setter
def crystal_size(self, xtal_size: CrystalSize):
self.__client.set(f"{self.__bl}:crystal_size", xtal_size.model_dump_json())
@property
def simple_input_parameters(self) -> SimpleStrategyInputModel | None:
tmp = self.__client.get(f"{self.__bl}:simple_input_params")
@@ -610,7 +607,14 @@ class BeamlineConfig:
tmp = self.__client.get(f"{self.__bl}:failed_mount_count")
if tmp is None:
return 0
return tmp
try:
return int(tmp)
except (TypeError, ValueError):
logger.warning(
"Failed Mount Count is not an integer, resetting to 0.",
extra={"beamline": self.__bl, "tmp": tmp},
)
return 0
@failed_mount_count.setter
def failed_mount_count(self, count:int):
+9 -1
View File
@@ -7,7 +7,7 @@ import cv2
import urllib3
import uvicorn
from aare.common.coordinate import SmargonCoordinate, Coordinate
from aare.common.error_codes import export_error_codes
from aare.common.error_codes import export_error_codes, export_error_codes_grouped
from aare.common.logger_config import setup_logger
from aare.common.models import SampleShortInfo, DAQStatusModel, BeamlineStateEnum, BeamlineSettingsModel, \
SampleShortInfoList, SessionStatus, SampleCameraSettings, AutofocusSettings, TokenData, \
@@ -62,6 +62,14 @@ async def meta_error_codes() -> dict[str, str]:
"""
return export_error_codes()
@app.get("/meta/error-codes/grouped")
async def meta_error_codes_grouped() -> dict[str, dict[str, str]]:
"""
Grouped registry of machine-readable error codes, by enum class.
Preferred for GUIs that want multiple code categories without collisions.
"""
return export_error_codes_grouped()
@app.get("/status")
async def status(token: str = Depends(oauth2_scheme)) -> DAQStatusModel:
data = auth.parse_token(token)
@@ -0,0 +1,380 @@
from __future__ import annotations
import json
from typing import Dict
from PySide6.QtCore import Qt, Slot
from PySide6.QtGui import QGuiApplication
from PySide6.QtWidgets import (
QDialog,
QVBoxLayout,
QHBoxLayout,
QLineEdit,
QPushButton,
QTabWidget,
QTableWidget,
QTableWidgetItem,
QTextEdit,
QLabel,
QWidget,
QCheckBox,
QDialogButtonBox,
QFormLayout,
QFrame,
)
from aare.common.error_codes import error_code_help
from aare.gui.threads.daq_worker import DAQWorker
class DeveloperHelpDialog(QDialog):
def __init__(self, *, daq: DAQWorker, is_staff: bool, parent=None):
super().__init__(parent)
self._daq = daq
self._is_staff = bool(is_staff)
self._codes: Dict[str, str] = {}
self._last_payload: dict = {}
self._freeze_payload: bool = False
self._always_highlight_last_error: bool = True
self.setWindowTitle("Developer / Help")
self.setMinimumSize(860, 560)
root = QVBoxLayout(self)
root.setContentsMargins(12, 12, 12, 12)
root.setSpacing(8)
# Compact banner (staff only)
self._banner = QLabel(self)
self._banner.setVisible(self._is_staff)
self._banner.setWordWrap(True)
self._banner.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self._banner.setStyleSheet(
"QLabel {"
" background: #f6f6f6;"
" border: 1px solid #d0d0d0;"
" border-radius: 6px;"
" padding: 6px 8px;"
"}"
)
root.addWidget(self._banner)
# Top controls
top = QHBoxLayout()
top.setSpacing(8)
root.addLayout(top)
top.addWidget(QLabel("Filter:", self))
self._filter = QLineEdit(self)
self._filter.setPlaceholderText("Type to filter (matches name, value, or help)…")
self._filter.setClearButtonEnabled(True)
self._filter.setMinimumHeight(28)
self._filter.setStyleSheet(
"QLineEdit {"
" background: white;"
" border: 1px solid #bdbdbd;"
" border-radius: 6px;"
" padding: 4px 8px;"
"}"
)
self._filter.textChanged.connect(self._apply_filter)
top.addWidget(self._filter, 1)
self._refresh_btn = QPushButton("Refresh", self)
self._refresh_btn.clicked.connect(self.refresh)
top.addWidget(self._refresh_btn)
self._freeze_cb = QCheckBox("Freeze payload", self)
self._freeze_cb.setVisible(self._is_staff)
self._freeze_cb.toggled.connect(self._set_freeze_payload)
top.addWidget(self._freeze_cb)
self._highlight_cb = QCheckBox("Always highlight last error code", self)
self._highlight_cb.setVisible(self._is_staff)
self._highlight_cb.setChecked(True)
self._highlight_cb.toggled.connect(self._set_always_highlight)
top.addWidget(self._highlight_cb)
self._copy_selected_btn = QPushButton("Copy code", self)
self._copy_selected_btn.clicked.connect(self._copy_selected_code)
top.addWidget(self._copy_selected_btn)
self._copy_all_btn = QPushButton("Copy all (filtered)", self)
self._copy_all_btn.clicked.connect(self._copy_all_filtered)
top.addWidget(self._copy_all_btn)
self._copy_payload_btn = QPushButton("Copy payload", self)
self._copy_payload_btn.setVisible(self._is_staff)
self._copy_payload_btn.clicked.connect(self._copy_payload)
top.addWidget(self._copy_payload_btn)
# Tabs
self._tabs = QTabWidget(self)
root.addWidget(self._tabs, 1)
# Tab: error codes + details pane
self._codes_table = QTableWidget(self)
self._codes_table.setColumnCount(1)
self._codes_table.setHorizontalHeaderLabels(["Name"])
self._codes_table.setSortingEnabled(True)
self._codes_table.setEditTriggers(QTableWidget.EditTrigger.NoEditTriggers)
self._codes_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows)
self._codes_table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection)
self._codes_table.itemSelectionChanged.connect(self._update_code_details)
self._codes_table.horizontalHeader().setStretchLastSection(True)
# Details panel (pretty form layout)
self._details_frame = QFrame(self)
self._details_frame.setFrameShape(QFrame.Shape.StyledPanel)
self._details_frame.setStyleSheet(
"QFrame {"
" background: #fafafa;"
" border: 1px solid #d0d0d0;"
" border-radius: 6px;"
"}"
)
details_layout = QVBoxLayout(self._details_frame)
details_layout.setContentsMargins(10, 10, 10, 10)
details_layout.setSpacing(8)
form = QFormLayout()
form.setLabelAlignment(Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter)
details_layout.addLayout(form)
self._detail_name = QLabel("-", self)
self._detail_name.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
form.addRow("Name:", self._detail_name)
value_row = QHBoxLayout()
value_row.setSpacing(8)
self._detail_value = QLabel("-", self)
self._detail_value.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self._copy_value_btn = QPushButton("Copy value", self)
self._copy_value_btn.clicked.connect(self._copy_selected_value)
value_row.addWidget(self._detail_value, 1)
value_row.addWidget(self._copy_value_btn, 0)
value_row_widget = QWidget(self)
value_row_widget.setLayout(value_row)
form.addRow("Value:", value_row_widget)
self._detail_help = QLabel("-", self)
self._detail_help.setWordWrap(True)
self._detail_help.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
self._detail_help.setStyleSheet(
"QLabel {"
" background: white;"
" border: 1px solid #e0e0e0;"
" border-radius: 6px;"
" padding: 8px;"
"}"
)
details_layout.addWidget(QLabel("Help:", self))
details_layout.addWidget(self._detail_help, 1)
codes_container = QWidget(self)
codes_layout = QVBoxLayout(codes_container)
codes_layout.setContentsMargins(0, 0, 0, 0)
codes_layout.setSpacing(8)
codes_layout.addWidget(self._codes_table, 1)
codes_layout.addWidget(self._details_frame, 0)
self._tabs.addTab(codes_container, "Error codes")
# Tab: last error payload (staff only)
self._payload_text = QTextEdit(self)
self._payload_text.setReadOnly(True)
self._payload_text.setLineWrapMode(QTextEdit.LineWrapMode.NoWrap)
if self._is_staff:
payload_container = QWidget(self)
payload_layout = QVBoxLayout(payload_container)
payload_layout.setContentsMargins(0, 0, 0, 0)
payload_layout.addWidget(self._payload_text, 1)
self._tabs.addTab(payload_container, "Last error payload")
else:
self._payload_text.setPlainText("Hidden (staff only).")
# Bottom button box
buttons = QDialogButtonBox(QDialogButtonBox.StandardButton.Close, parent=self)
buttons.rejected.connect(self.reject)
buttons.accepted.connect(self.accept)
root.addWidget(buttons)
# Wire signals
self._daq.error_codes_loaded.connect(self.set_error_codes)
self._daq.last_error_payload_changed.connect(self.set_last_error_payload)
self._update_banner()
self._update_code_details()
@Slot(bool)
def _set_freeze_payload(self, enabled: bool) -> None:
self._freeze_payload = bool(enabled)
@Slot(bool)
def _set_always_highlight(self, enabled: bool) -> None:
self._always_highlight_last_error = bool(enabled)
@Slot()
def refresh(self) -> None:
self._daq.get_error_codes()
self.set_last_error_payload(self._daq.get_last_error_payload())
@Slot(dict)
def set_error_codes(self, codes: dict) -> None:
self._codes = {str(k): str(v) for k, v in (codes or {}).items()}
self._apply_filter()
@Slot(dict)
def set_last_error_payload(self, payload: dict) -> None:
if not self._is_staff:
return
if self._freeze_payload:
return
self._last_payload = payload or {}
pretty = json.dumps(self._last_payload, indent=2, sort_keys=True, default=str)
self._payload_text.setPlainText(pretty)
self._update_banner()
self._select_code_from_last_error()
def _code_name_for_value(self, value: str) -> str | None:
for k, v in self._codes.items():
if v == value:
return k
return None
def _extract_code_message(self) -> tuple[str | None, str | None]:
body = (self._last_payload or {}).get("body_json")
if isinstance(body, dict):
code = body.get("code")
msg = body.get("message")
return (str(code) if code is not None else None, str(msg) if msg is not None else None)
return (None, None)
def _update_banner(self) -> None:
if not self._is_staff:
return
code, msg = self._extract_code_message()
if code or msg:
parts = []
if code:
parts.append(f"Last error: {code}")
if msg:
parts.append(msg)
self._banner.setText("".join(parts))
else:
self._banner.setText("Last error: (none captured yet)")
def _code_name_for_value(self, value: str) -> str | None:
for name, v in (self._codes or {}).items():
if str(v) == str(value):
return str(name)
return None
def _select_code_by_name(self, name: str) -> None:
if not name:
return
for row in range(self._codes_table.rowCount()):
item = self._codes_table.item(row, 0)
if item and item.text() == name:
self._codes_table.setCurrentCell(row, 0)
self._codes_table.scrollToItem(item)
return
def _select_code_from_last_error(self) -> None:
if not self._codes:
return
code_value, _ = self._extract_code_message()
if not code_value:
return
name = self._code_name_for_value(code_value)
if name:
self._select_code_by_name(name)
@Slot()
def _apply_filter(self) -> None:
term = (self._filter.text() or "").strip().lower()
items = sorted(self._codes.items(), key=lambda kv: kv[0])
if term:
def _match(name: str, value: str) -> bool:
h = error_code_help(value) or ""
return term in f"{name}\n{value}\n{h}".lower()
items = [(k, v) for (k, v) in items if _match(k, v)]
self._codes_table.setRowCount(len(items))
for row, (k, v) in enumerate(items):
item = QTableWidgetItem(k)
item.setFlags(item.flags() & ~Qt.ItemFlag.ItemIsEditable)
self._codes_table.setItem(row, 0, item)
self._codes_table.resizeColumnsToContents()
# Keep selection sensible (dont resize columns here; avoids jitter)
if self._codes_table.rowCount() > 0 and self._codes_table.currentRow() < 0:
self._codes_table.setCurrentCell(0, 0)
self._update_code_details()
# If enabled, re-try selection after the table content changes
if self._is_staff and self._always_highlight_last_error and not self._freeze_payload:
self._select_code_from_last_error()
def _selected_code(self) -> tuple[str | None, str | None]:
row = self._codes_table.currentRow()
if row < 0:
return (None, None)
name_item = self._codes_table.item(row, 0)
if not name_item:
return (None, None)
name = name_item.text()
value = self._codes.get(name)
return (name, value)
@Slot()
def _update_code_details(self) -> None:
name, value = self._selected_code()
if not name or not value:
self._detail_name.setText("-")
self._detail_value.setText("-")
self._detail_help.setText("Select an error code to see details.")
self._copy_value_btn.setEnabled(False)
return
self._detail_name.setText(name)
self._detail_value.setText(value)
self._detail_help.setText(error_code_help(value) or "(no help text defined yet)")
self._copy_value_btn.setEnabled(True)
@Slot()
def _copy_selected_code(self) -> None:
name, value = self._selected_code()
if not name or not value:
QGuiApplication.clipboard().setText("")
return
QGuiApplication.clipboard().setText(f"{name}={value}")
@Slot()
def _copy_selected_value(self) -> None:
_, value = self._selected_code()
QGuiApplication.clipboard().setText(value or "")
@Slot()
def _copy_all_filtered(self) -> None:
rows = self._codes_table.rowCount()
out = {}
for r in range(rows):
name = self._codes_table.item(r, 0).text()
out[name] = self._codes.get(name, "")
QGuiApplication.clipboard().setText(json.dumps(out, indent=2, sort_keys=True))
@Slot()
def _copy_payload(self) -> None:
if not self._is_staff:
return
QGuiApplication.clipboard().setText(self._payload_text.toPlainText())
+115 -8
View File
@@ -8,6 +8,7 @@ from PySide6.QtNetwork import QNetworkAccessManager, QNetworkRequest, QNetworkRe
from jfjoch_client import ScanResult, ScanResultImagesInner
from aare.common.coordinate import SmargonCoordinate, Coordinate
from aare.common.error_codes import export_error_codes
from aare.common.models import DAQStatusModel, SampleShortInfoList, SampleShortInfo, SampleCameraSettings, \
AutofocusSettings, SimpleScanParameters, FluorescenceSpectrumParameterModel, FluorescenceSpectrumOutputModel
from aare.common.raster_grid import RasterGridRequest, CompletedRasterGrid
@@ -37,6 +38,9 @@ class DAQWorker(QObject):
fluorimeter_update = Signal(list, list, int)
fluorimeter_spectrum_update = Signal(FluorescenceSpectrumOutputModel)
error_codes_loaded = Signal(dict)
last_error_payload_changed = Signal(dict)
def __init__(self, base_url: str | None, token: str, parent=None):
super().__init__(parent)
self.__token = token
@@ -55,6 +59,15 @@ class DAQWorker(QObject):
self._status_request_min_interval = 10.0
self._last_status_error = None
self._last_error_payload: dict = {}
def get_last_error_payload(self) -> dict:
return dict(self._last_error_payload or {})
def _set_last_error_payload(self, payload: dict) -> None:
self._last_error_payload = payload or {}
self.last_error_payload_changed.emit(self.get_last_error_payload())
@Slot()
def regular_update(self):
if self.__counter % SPREADHSEET_FREQUENCY == 0:
@@ -125,18 +138,44 @@ class DAQWorker(QObject):
if reply.error() != QNetworkReply.NetworkError.NoError:
status = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)
err_details = reply.errorString()
raw_body = ""
body_json = None
try:
response_body = reply.readAll().data().decode("utf-8")
if response_body:
body_json = json.loads(response_body)
if "detail" in body_json:
err_details = body_json["detail"]
raw_body = reply.readAll().data().decode("utf-8")
if raw_body:
body_json = json.loads(raw_body)
if isinstance(body_json, dict):
if "detail" in body_json:
err_details = body_json["detail"]
elif "message" in body_json:
err_details = body_json["message"]
else:
err_details = raw_body
else:
err_details = response_body
err_details = raw_body
except Exception:
pass
if status== 401:
try:
url = reply.request().url().toString()
except Exception:
url = ""
net_err = reply.error()
net_err_name = getattr(net_err, "name", None)
net_err_value = getattr(net_err, "value", None)
self._set_last_error_payload({
"url": url,
"http_status": int(status) if status is not None else None,
"network_error": net_err_name or str(net_err),
"network_error_value": int(net_err_value) if isinstance(net_err_value, int) else None,
"error_string": str(reply.errorString()),
"body_raw": raw_body,
"body_json": body_json,
})
if status == 401:
now = time.monotonic()
if now - self._last_auth_error_log_ts > self._auth_error_min_interval:
logger.error(f"{err_details}: baton taken by another user")
@@ -147,6 +186,7 @@ class DAQWorker(QObject):
else:
logger.error(f"{err_details}")
self.http_error.emit(err_details)
reply.deleteLater()
def generic_post(self, url: str, body: str = ""):
@@ -658,4 +698,71 @@ class DAQWorker(QObject):
status = obj.get("status", -1)
self.fluorimeter_update.emit(data, bkg, status)
except Exception as e:
logger.error(f"SSE parse error: {e}")
logger.error(f"SSE parse error: {e}")
@staticmethod
def _flatten_error_codes_payload(obj: dict) -> dict[str, str]:
"""
Accept either:
- flat: {"INVALID_TOKEN": "INVALID_TOKEN"}
- grouped: {"AuthErrorCode": {"INVALID_TOKEN": "INVALID_TOKEN"}, "DAQErrorCode": {...}}
Output is always flat strings, using 'Group.KEY' for grouped input.
"""
out: dict[str, str] = {}
for k, v in (obj or {}).items():
if isinstance(v, dict):
group = str(k)
for kk, vv in v.items():
out[f"{group}.{str(kk)}"] = str(vv)
else:
out[str(k)] = str(v)
return out
@Slot()
def get_error_codes(self) -> None:
"""
Fetch server error codes registry for developer/help UI.
Emits error_codes_loaded(dict).
"""
if self.__base_url is None:
self.error_codes_loaded.emit(export_error_codes())
return
# Prefer grouped (newer servers), but we'll gracefully fall back if missing.
request = QNetworkRequest(QUrl(f"{self.__base_url}/meta/error-codes/grouped"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_error_codes_response(reply))
def _retry_error_codes_legacy(self) -> None:
request = QNetworkRequest(QUrl(f"{self.__base_url}/meta/error-codes"))
request.setRawHeader(b"Authorization", f"Bearer {self.__token}".encode("utf-8"))
reply = self.__net_manager.get(request)
reply.finished.connect(lambda: self._handle_error_codes_response(reply))
@Slot(QNetworkReply)
def _handle_error_codes_response(self, reply: QNetworkReply) -> None:
try:
status = reply.attribute(QNetworkRequest.Attribute.HttpStatusCodeAttribute)
# If the grouped endpoint isn't available on this server, fall back to legacy.
try:
url = reply.request().url().toString()
except Exception:
url = ""
if int(status) == 404 and url.endswith("/meta/error-codes/grouped"):
reply.deleteLater()
self._retry_error_codes_legacy()
return
payload = self.handle_response(reply)
obj = json.loads(payload) if payload else {}
if not isinstance(obj, dict):
raise RuntimeError("Invalid error-codes payload (expected JSON object)")
out = self._flatten_error_codes_payload(obj)
self.error_codes_loaded.emit(out)
except Exception as e:
logger.error(f"Failed to load error codes: {e}")
self.http_error.emit(str(e))