Coverage for src / lexigram / admin / controllers / widgets.py: 17%
256 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Widget controller — routes HTMX widget requests to contributors."""
3from __future__ import annotations
5import inspect
6from typing import Any, cast
8from starlette.requests import Request
9from starlette.routing import Route
11from lexigram.admin.auth.protocols import (
12 AdminAuditLogServiceProtocol,
13 AdminCsrfServiceProtocol,
14)
15from lexigram.admin.auth.types import AdminSecurityEventType
16from lexigram.admin.dashboard.widget_types import ConfigField
17from lexigram.admin.params import parse_widget_params
18from lexigram.contracts.admin.protocols import AdminContributorRegistryProtocol
19from lexigram.contracts.web import get, post
20from lexigram.di.decorators import inject
21from lexigram.logging import get_logger
22from lexigram.ui import el
24logger = get_logger(__name__)
26_REQUIRED_PERMISSIONS = frozenset({"admin.settings.edit"})
29@inject
30class WidgetController:
31 """Routes HTMX widget/health requests to the appropriate contributor.
33 All dependencies are constructor-injected. No service locator.
35 Args:
36 registry: The admin contributor registry.
37 """
39 prefix = ""
41 def __init__(
42 self,
43 registry: AdminContributorRegistryProtocol,
44 audit_service: AdminAuditLogServiceProtocol | None = None,
45 csrf_service: AdminCsrfServiceProtocol | None = None,
46 ) -> None:
47 self._registry = registry
48 self._settings_service: Any = None
49 self._resolver: Any = None
50 self._audit_service = audit_service
51 self._csrf_service = csrf_service
53 def _get_csrf_token(self, request: Request) -> str | None:
54 """Resolve the CSRF token for form rendering, if available."""
55 if not self._csrf_service:
56 return None
57 try:
58 session = getattr(request, "session", {})
59 session_id: str = session.get("admin_user_id", "")
60 return self._csrf_service.generate_token(session_id)
61 except Exception: # noqa: BLE001 — non-fatal for form rendering
62 return None
64 # -- helpers --
66 def _user_has_edit_permission(self, request: Request) -> bool:
67 """Check whether the requesting user may mutate widget prefs.
69 Superadmin bypasses permission gating so accounts created with an
70 empty permission set (e.g. via the setup wizard) can still manage
71 dashboard widgets.
72 """
73 user = getattr(getattr(request, "state", None), "user", None)
74 if user and "superadmin" in (getattr(user, "roles", None) or ()):
75 return True
76 permissions = frozenset(getattr(user, "permissions", None) or ())
77 return permissions.issuperset(_REQUIRED_PERMISSIONS)
79 async def _audit(
80 self,
81 request: Request,
82 success: bool = True,
83 event_type: AdminSecurityEventType = AdminSecurityEventType.SETTINGS_UPDATED,
84 **metadata: Any,
85 ) -> None:
86 """Append a widget-prefs change to the security audit log, best-effort."""
87 if not self._audit_service:
88 return
89 try:
90 client = getattr(request, "client", None)
91 await self._audit_service.log_event(
92 event_type=event_type,
93 ip_address=getattr(client, "host", "unknown"),
94 user_agent=request.headers.get("user-agent", "") or "",
95 success=success,
96 metadata=metadata,
97 )
98 except Exception: # noqa: BLE001 — audit failures must not break saves
99 logger.warning("widgets.audit_failed", **metadata)
101 def get_routes(self) -> list[Any]:
102 """Extract decorated routes from this controller instance."""
103 routes = []
105 for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
106 if hasattr(method, "_route_config"):
107 config = method._route_config
108 method_params = list(inspect.signature(method).parameters.keys())
110 async def starlette_handler(
111 request: Request, m=method, params=method_params
112 ) -> Any:
113 kwargs = {"request": request}
114 for param in params:
115 if param == "request":
116 continue
117 if param in request.path_params:
118 kwargs[param] = request.path_params[param]
119 return await m(**kwargs)
121 base_path = getattr(self, "prefix", "").rstrip("/")
122 route_path = config["path"]
123 if not route_path.startswith("/"):
124 route_path = f"/{route_path}"
126 if route_path == "/" and base_path:
127 full_path = base_path
128 else:
129 full_path = f"{base_path}{route_path}" if base_path else route_path
131 routes.append(
132 Route(
133 full_path,
134 endpoint=starlette_handler,
135 methods=[config["method"]],
136 name=config.get("name") or f"admin_widget_{name}",
137 ),
138 )
140 return routes
142 @get("/{contributor_id}/widgets/{widget_name}")
143 async def render_widget(
144 self,
145 request: Request,
146 contributor_id: str,
147 widget_name: str,
148 ) -> object:
149 """Render a widget fragment for the HTMX dashboard.
151 Returns an inline error card when the widget is not found or fails
152 to render, rather than returning an HTTP error status. This lets
153 the dashboard grid maintain its layout even when individual widgets
154 fail.
155 """
156 from starlette.responses import HTMLResponse
158 contributor = self._registry.get(contributor_id)
159 if contributor is None:
160 logger.warning(
161 "widget_contributor_not_found",
162 contributor_id=contributor_id,
163 widget_name=widget_name,
164 )
165 return HTMLResponse(
166 self._render_error_card(
167 f"Contributor '{contributor_id}' not found",
168 contributor_id=contributor_id,
169 widget_name=widget_name,
170 ),
171 status_code=200,
172 )
174 params = parse_widget_params(dict(request.query_params))
176 if self._settings_service:
177 full_name = f"{contributor_id}.{widget_name}"
178 prefs = await self._settings_service.get_widget_prefs("default", "default")
179 cfg = prefs.get("configs", {}).get(full_name, {})
180 if "time_window_minutes" in cfg:
181 from lexigram.contracts.admin.types import WidgetParams
183 params = WidgetParams(
184 page=params.page,
185 page_size=params.page_size,
186 time_window_minutes=int(cfg["time_window_minutes"]),
187 raw=params.raw,
188 )
190 result = await contributor.render_widget(
191 widget_name, params, resolver=self._resolver
192 )
194 if result.is_ok():
195 vm = result.unwrap()
196 return HTMLResponse(self._wrap_widget_body(vm.body, vm.title, vm.error))
198 error = result.unwrap_err()
199 logger.error(
200 "widget_render_failed",
201 contributor_id=contributor_id,
202 widget_name=widget_name,
203 error=str(error),
204 )
205 return HTMLResponse(
206 self._render_error_card(
207 str(error),
208 contributor_id=contributor_id,
209 widget_name=widget_name,
210 ),
211 status_code=200,
212 )
214 @staticmethod
215 def _wrap_widget_body(
216 body: str,
217 title: str | None = None,
218 error: str | None = None,
219 ) -> str:
220 """Wrap widget body in a container with optional error banner."""
221 from lexigram.ui.core.base import el, render_to_string
223 children: list[object] = []
224 if title:
225 children.append(
226 el(
227 "div",
228 title,
229 class_="widget-title text-xs font-semibold text-muted-foreground mb-1",
230 )
231 )
232 if error:
233 children.append(
234 el(
235 "div",
236 error,
237 class_="text-xs text-destructive bg-destructive/10 rounded px-2 py-1 mb-2",
238 )
239 )
240 children.append(el("div", body, class_="widget-content"))
241 return render_to_string(el("div", *children, class_="widget-body-container"))
243 @staticmethod
244 def _render_error_card(
245 message: str,
246 contributor_id: str | None = None,
247 widget_name: str | None = None,
248 ) -> str:
249 """Render an inline error card for failed widget rendering."""
250 from lexigram.ui.core.base import el, render_to_string
252 data_attrs: dict[str, str] = {}
253 if contributor_id:
254 data_attrs["data-contributor-id"] = contributor_id
255 if widget_name:
256 data_attrs["data-widget-name"] = widget_name
258 return render_to_string(
259 el(
260 "div",
261 el(
262 "div",
263 class_="text-destructive text-lg mb-1",
264 ),
265 el(
266 "p",
267 message,
268 class_="text-sm text-muted-foreground",
269 ),
270 class_="widget-error-card bg-destructive/10 border border-destructive/30 rounded-lg p-4 text-center",
271 **data_attrs,
272 )
273 )
275 @get("/{contributor_id}/health/{check_name}")
276 async def render_health_check(
277 self,
278 request: Request,
279 contributor_id: str,
280 check_name: str,
281 ) -> object:
282 """Render a health check fragment for the HTMX dashboard."""
283 from starlette.responses import HTMLResponse, Response
285 contributor = self._registry.get(contributor_id)
286 if contributor is None:
287 return Response(
288 content=f"Contributor '{contributor_id}' not found",
289 status_code=404,
290 )
292 result = await contributor.render_health_check(check_name)
294 if result.is_ok():
295 return HTMLResponse(result.unwrap())
297 error = result.unwrap_err()
298 return Response(content=str(error), status_code=422)
300 @get("/core/widgets/{name}/config")
301 async def widget_config_popup(
302 self,
303 request: Request,
304 name: str,
305 ) -> object:
306 """Render config popup for a widget."""
307 from starlette.responses import HTMLResponse
309 from lexigram.admin.dashboard.widgets import render_widget_config_popup
311 tenant_id = "default"
312 user_id = "default"
314 widget_def = None
315 contributor = None
316 for c in self._registry.get_all():
317 for wdef in c.get_dashboard_widgets():
318 if wdef.name == name:
319 widget_def = wdef
320 contributor = c
321 break
322 if widget_def:
323 break
325 if widget_def is None:
326 return HTMLResponse("Widget not found", status_code=404)
328 schema: list[ConfigField] = getattr(
329 contributor, "get_widget_config_schema", lambda _: []
330 )(name)
332 prefs = (
333 await self._settings_service.get_widget_prefs(tenant_id, user_id)
334 if self._settings_service
335 else {}
336 )
337 enabled = "enabled" not in prefs or name in prefs.get("enabled", [])
338 cfg = prefs.get("configs", {}).get(name, {})
340 html = render_widget_config_popup(name, widget_def.title, schema, cfg, enabled)
341 return HTMLResponse(html)
343 @post("/core/widgets/config")
344 async def save_widget_config(self, request: Request) -> object:
345 """Save a single widget's configuration."""
346 from starlette.responses import HTMLResponse
348 if not self._user_has_edit_permission(request):
349 await self._audit(
350 request,
351 success=False,
352 event_type=AdminSecurityEventType.PERMISSION_DENIED,
353 reason="permission_denied",
354 route="save_widget_config",
355 )
356 return HTMLResponse("Permission denied", status_code=403)
358 tenant_id = "default"
359 user_id = "default"
361 data = request.scope.get("admin_form_data")
362 if data is None:
363 data = await request.form()
364 widget_name = data.get("widget_name")
365 enabled = "enabled" in data
366 params = {
367 k.removeprefix("param_"): v
368 for k, v in data.items()
369 if k.startswith("param_")
370 }
372 prefs = (
373 await self._settings_service.get_widget_prefs(tenant_id, user_id)
374 if self._settings_service
375 else {}
376 )
377 enabled_list = prefs.get("enabled", [])
378 if enabled and widget_name not in enabled_list:
379 enabled_list.append(widget_name)
380 elif not enabled and widget_name in enabled_list:
381 enabled_list.remove(widget_name)
383 configs = prefs.get("configs", {})
384 if params:
385 configs[widget_name] = params
386 elif widget_name in configs:
387 del configs[widget_name]
389 prefs["enabled"] = enabled_list
390 prefs["configs"] = configs
391 if self._settings_service:
392 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
393 await self._audit(
394 request,
395 widget_name=widget_name or "",
396 kind="widget_config",
397 )
398 return HTMLResponse("", status_code=204)
400 @post("/core/widgets/reorder")
401 async def reorder_widgets(self, request: Request) -> object:
402 """Save widget order after drag-and-drop."""
403 from starlette.responses import HTMLResponse
405 if not self._user_has_edit_permission(request):
406 await self._audit(
407 request,
408 success=False,
409 event_type=AdminSecurityEventType.PERMISSION_DENIED,
410 reason="permission_denied",
411 route="reorder_widgets",
412 )
413 return HTMLResponse("Permission denied", status_code=403)
415 tenant_id = "default"
416 user_id = "default"
418 data = await request.json()
419 order_list = data.get("order", [])
420 prefs = (
421 await self._settings_service.get_widget_prefs(tenant_id, user_id)
422 if self._settings_service
423 else {}
424 )
425 prefs["order"] = {name: idx for idx, name in enumerate(order_list)}
426 if self._settings_service:
427 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
428 await self._audit(request, kind="widget_reorder")
429 return HTMLResponse("", status_code=204)
431 @get("/core/widgets/customize")
432 async def customize_all_widgets(self, request: Request) -> object:
433 """Render full dashboard customization panel with all widgets."""
434 from starlette.responses import HTMLResponse
436 tenant_id = "default"
437 user_id = "default"
439 prefs = (
440 await self._settings_service.get_widget_prefs(tenant_id, user_id)
441 if self._settings_service
442 else {}
443 )
444 enabled_list = prefs.get("enabled", [])
445 has_explicit_prefs = "enabled" in prefs
447 grouped: dict[str, tuple[str, list[Any]]] = {}
448 for contributor in self._registry.get_all():
449 widgets = list(contributor.get_dashboard_widgets())
450 if not widgets:
451 continue
452 toggles: list[Any] = []
453 for wdef in widgets:
454 name = wdef.name
455 enabled = name in enabled_list
456 if not has_explicit_prefs:
457 # No saved prefs yet — dashboard shows everything by default,
458 # so the form must render every widget as enabled.
459 enabled = True
460 toggles.append(
461 el(
462 "label",
463 el(
464 "input",
465 type_="checkbox",
466 name=f"enabled_{name}",
467 value="1",
468 checked="checked" if enabled else None,
469 ),
470 el(
471 "span",
472 wdef.title,
473 class_="truncate text-sm font-medium",
474 ),
475 class_=(
476 "flex items-center gap-2 rounded-lg border border-border "
477 "bg-card px-3 py-2 cursor-pointer select-none "
478 "hover:bg-muted/50 transition-colors"
479 ),
480 )
481 )
482 label = getattr(contributor, "display_name", "") or contributor.name
483 grouped[contributor.name] = (label, toggles)
485 sections: list[Any] = []
486 for label, toggles in grouped.values():
487 sections.append(
488 el(
489 "div",
490 el(
491 "h3",
492 label,
493 class_=(
494 "text-xs font-semibold uppercase tracking-wider "
495 "text-muted-foreground"
496 ),
497 ),
498 el("div", class_="mt-1.5 border-t border-border"),
499 el(
500 "div",
501 *toggles,
502 class_="mt-3 grid grid-cols-2 gap-2",
503 ),
504 )
505 )
507 csrf_token = self._get_csrf_token(request)
509 from lexigram.admin.ui.organisms.admin_slide_over import (
510 render_slide_over_fragment,
511 )
513 form = el(
514 "form",
515 el("input", type_="hidden", name="csrf_token", value=csrf_token or ""),
516 *sections,
517 id="widget-customize-form",
518 **{
519 "hx-post": "/admin/core/widgets/customize/save",
520 "hx-swap": "none",
521 "hx-on:htmx:after-request": "if(event.detail.successful){window.location.reload();}",
522 },
523 class_="space-y-6",
524 )
526 return HTMLResponse(
527 render_slide_over_fragment(
528 title="Customize Dashboard",
529 subtitle="Enable/disable widgets and configure their parameters.",
530 content=form,
531 size="xl",
532 footer=[
533 el(
534 "button",
535 "Cancel",
536 type_="button",
537 **{"x-on:click": "open = false"},
538 class_="inline-flex items-center rounded-lg px-4 py-2 text-sm font-medium text-foreground bg-card border border-border hover:bg-muted transition-colors",
539 ),
540 el(
541 "button",
542 "Save All Changes",
543 type_="submit",
544 form="widget-customize-form",
545 class_="inline-flex items-center rounded-lg px-4 py-2 text-sm font-medium text-white bg-primary hover:bg-primary/90 transition-colors",
546 ),
547 ],
548 )
549 )
551 @post("/core/widgets/customize/save")
552 async def save_all_widget_configs(self, request: Request) -> object:
553 """Save all widget configurations from the customize panel."""
554 from starlette.responses import HTMLResponse
556 if not self._user_has_edit_permission(request):
557 await self._audit(
558 request,
559 success=False,
560 event_type=AdminSecurityEventType.PERMISSION_DENIED,
561 reason="permission_denied",
562 route="save_all_widget_configs",
563 )
564 return HTMLResponse("Permission denied", status_code=403)
566 tenant_id = "default"
567 user_id = "default"
569 data = request.scope.get("admin_form_data")
570 if data is None:
571 data = await request.form()
573 enabled_list: list[str] = []
574 configs: dict[str, dict[str, str]] = {}
575 all_widget_names: list[str] = []
577 for contributor in self._registry.get_all():
578 for wdef in contributor.get_dashboard_widgets():
579 all_widget_names.append(wdef.name)
581 for key, val in data.items():
582 if key.startswith("enabled_"):
583 wname = key.removeprefix("enabled_")
584 if wname in all_widget_names and val:
585 enabled_list.append(wname)
586 elif key.startswith("param__"):
587 rest = key.removeprefix("param__")
588 wname, pname = rest.split("__", 1)
589 configs.setdefault(wname, {})[pname] = cast("str", val)
591 existing = (
592 await self._settings_service.get_widget_prefs(tenant_id, user_id)
593 if self._settings_service
594 else {}
595 )
596 prefs = {
597 "enabled": enabled_list,
598 "configs": configs,
599 "order": existing.get("order", {}),
600 }
601 if self._settings_service:
602 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
603 await self._audit(
604 request,
605 kind="widget_customize",
606 widget_count=len(all_widget_names),
607 )
608 return HTMLResponse("", status_code=204)
611__all__ = ["WidgetController"]