Coverage for src/lexigram/admin/controllers/widgets.py: 79%
289 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 16:30 +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.config import AdminRbacConfig
17from lexigram.admin.dashboard.content_renderer import render_content
18from lexigram.admin.dashboard.widget_types import ConfigField
19from lexigram.admin.multitenancy.adapter import resolve_tenant_id
20from lexigram.admin.params import parse_widget_params
21from lexigram.admin.rbac.super_admin import is_super_admin
22from lexigram.contracts.admin.protocols import AdminContributorRegistryProtocol
23from lexigram.contracts.admin.types import WidgetContent
24from lexigram.contracts.web import get, post
25from lexigram.di.decorators import inject
26from lexigram.logging import get_logger
27from lexigram.ui import el
29logger = get_logger(__name__)
31_REQUIRED_PERMISSIONS = frozenset({"admin.settings.edit"})
34@inject
35class WidgetController:
36 """Routes HTMX widget/health requests to the appropriate contributor.
38 All dependencies are constructor-injected. No service locator.
40 Args:
41 registry: The admin contributor registry.
42 """
44 prefix = ""
46 def __init__(
47 self,
48 registry: AdminContributorRegistryProtocol,
49 audit_service: AdminAuditLogServiceProtocol | None = None,
50 csrf_service: AdminCsrfServiceProtocol | None = None,
51 rbac_config: AdminRbacConfig | None = None,
52 ) -> None:
53 self._registry = registry
54 self._settings_service: Any = None
55 self._resolver: Any = None
56 self._audit_service = audit_service
57 self._csrf_service = csrf_service
58 self._rbac_config = rbac_config
60 def _get_csrf_token(self, request: Request) -> str | None:
61 """Resolve the CSRF token for form rendering, if available."""
62 if not self._csrf_service:
63 return None
64 try:
65 session = getattr(request, "session", {})
66 session_id: str = session.get("admin_user_id", "")
67 return self._csrf_service.generate_token(session_id)
68 except Exception: # noqa: BLE001 — non-fatal for form rendering
69 return None
71 # -- helpers --
73 def _user_has_edit_permission(self, request: Request) -> bool:
74 """Check whether the requesting user may mutate widget prefs.
76 Superadmin bypasses permission gating so accounts created with an
77 empty permission set (e.g. via the setup wizard) can still manage
78 dashboard widgets.
79 """
80 user = getattr(getattr(request, "state", None), "user", None)
81 if self._user_is_superadmin(request):
82 return True
83 permissions = frozenset(getattr(user, "permissions", None) or ())
84 return permissions.issuperset(_REQUIRED_PERMISSIONS)
86 def _user_permissions(self, request: Request) -> frozenset[str]:
87 """Return the requesting user's permission set (empty when absent)."""
88 user = getattr(getattr(request, "state", None), "user", None)
89 return frozenset(getattr(user, "permissions", None) or ())
91 def _user_is_superadmin(self, request: Request) -> bool:
92 """Return True when the requesting user holds the superadmin role."""
93 role = (self._rbac_config or AdminRbacConfig()).super_admin_role
94 user = getattr(getattr(request, "state", None), "user", None)
95 return bool(user) and is_super_admin(user, role)
97 def _has_required_permission(self, request: Request, required: str | None) -> bool:
98 """Check *required* against the requesting user; superadmin bypasses."""
99 if not required:
100 return True
101 if self._user_is_superadmin(request):
102 return True
103 return required in self._user_permissions(request)
105 async def _audit(
106 self,
107 request: Request,
108 success: bool = True,
109 event_type: AdminSecurityEventType = AdminSecurityEventType.SETTINGS_UPDATED,
110 **metadata: Any,
111 ) -> None:
112 """Append a widget-prefs change to the security audit log, best-effort."""
113 if not self._audit_service:
114 return
115 try:
116 client = getattr(request, "client", None)
117 await self._audit_service.log_event(
118 event_type=event_type,
119 ip_address=getattr(client, "host", "unknown"),
120 user_agent=request.headers.get("user-agent", "") or "",
121 success=success,
122 metadata=metadata,
123 )
124 except Exception: # noqa: BLE001 — audit failures must not break saves
125 logger.warning("widgets.audit_failed", **metadata)
127 def get_routes(self) -> list[Any]:
128 """Extract decorated routes from this controller instance."""
129 routes = []
131 for name, method in inspect.getmembers(self, predicate=inspect.ismethod):
132 if hasattr(method, "_route_config"):
133 config = method._route_config
134 method_params = list(inspect.signature(method).parameters.keys())
136 async def starlette_handler(
137 request: Request, m=method, params=method_params
138 ) -> Any:
139 kwargs = {"request": request}
140 for param in params:
141 if param == "request":
142 continue
143 if param in request.path_params:
144 kwargs[param] = request.path_params[param]
145 return await m(**kwargs)
147 base_path = getattr(self, "prefix", "").rstrip("/")
148 route_path = config["path"]
149 if not route_path.startswith("/"):
150 route_path = f"/{route_path}"
152 if route_path == "/" and base_path:
153 full_path = base_path
154 else:
155 full_path = f"{base_path}{route_path}" if base_path else route_path
157 routes.append(
158 Route(
159 full_path,
160 endpoint=starlette_handler,
161 methods=[config["method"]],
162 name=config.get("name") or f"admin_widget_{name}",
163 ),
164 )
166 return routes
168 @get("/{contributor_id}/widgets/{widget_name}")
169 async def render_widget(
170 self,
171 request: Request,
172 contributor_id: str,
173 widget_name: str,
174 ) -> object:
175 """Render a widget fragment for the HTMX dashboard.
177 Returns an inline error card when the widget is not found or fails
178 to render, rather than returning an HTTP error status. This lets
179 the dashboard grid maintain its layout even when individual widgets
180 fail.
181 """
182 from starlette.responses import HTMLResponse
184 contributor = self._registry.get(contributor_id)
185 if contributor is None:
186 logger.warning(
187 "widget_contributor_not_found",
188 contributor_id=contributor_id,
189 widget_name=widget_name,
190 )
191 return HTMLResponse(
192 self._render_error_card(
193 f"Contributor '{contributor_id}' not found",
194 contributor_id=contributor_id,
195 widget_name=widget_name,
196 ),
197 status_code=200,
198 )
200 widget_def = next(
201 (w for w in contributor.get_dashboard_widgets() if w.name == widget_name),
202 None,
203 )
204 if widget_def is not None and not self._has_required_permission(
205 request, widget_def.permission
206 ):
207 return HTMLResponse(
208 self._render_error_card(
209 "You do not have permission to view this widget.",
210 contributor_id=contributor_id,
211 widget_name=widget_name,
212 ),
213 status_code=200,
214 )
216 params = parse_widget_params(dict(request.query_params))
218 if self._settings_service:
219 full_name = f"{contributor_id}.{widget_name}"
220 tenant_id = await resolve_tenant_id(request, default="default")
221 prefs = await self._settings_service.get_widget_prefs(tenant_id, "default")
222 cfg = prefs.get("configs", {}).get(full_name, {})
223 if "time_window_minutes" in cfg:
224 from lexigram.contracts.admin.types import WidgetParams
226 params = WidgetParams(
227 page=params.page,
228 page_size=params.page_size,
229 time_window_minutes=int(cfg["time_window_minutes"]),
230 raw=params.raw,
231 )
233 result = await contributor.render_widget(
234 widget_name, params, resolver=self._resolver
235 )
237 if result.is_ok():
238 vm = result.unwrap()
239 return HTMLResponse(self._wrap_widget_body(vm.content, vm.title, vm.error))
241 error = result.unwrap_err()
242 logger.error(
243 "widget_render_failed",
244 contributor_id=contributor_id,
245 widget_name=widget_name,
246 error=str(error),
247 )
248 return HTMLResponse(
249 self._render_error_card(
250 str(error),
251 contributor_id=contributor_id,
252 widget_name=widget_name,
253 ),
254 status_code=200,
255 )
257 @staticmethod
258 def _wrap_widget_body(
259 content: WidgetContent,
260 title: str | None = None,
261 error: str | None = None,
262 ) -> str:
263 """Wrap widget body in a container with optional error banner."""
264 from lexigram.ui.core.base import el, raw, render_to_string
266 inner = render_content(content)
267 children: list[object] = []
268 if title:
269 children.append(
270 el(
271 "div",
272 title,
273 class_="widget-title text-xs font-semibold text-muted-foreground mb-1",
274 )
275 )
276 if error:
277 children.append(
278 el(
279 "div",
280 error,
281 class_="text-xs text-destructive bg-destructive/10 rounded px-2 py-1 mb-2",
282 )
283 )
284 children.append(el("div", raw(inner), class_="widget-content"))
285 return render_to_string(el("div", *children, class_="widget-body-container"))
287 @staticmethod
288 def _render_error_card(
289 message: str,
290 contributor_id: str | None = None,
291 widget_name: str | None = None,
292 ) -> str:
293 """Render an inline error card for failed widget rendering."""
294 from lexigram.ui.core.base import el, render_to_string
296 data_attrs: dict[str, str] = {}
297 if contributor_id:
298 data_attrs["data-contributor-id"] = contributor_id
299 if widget_name:
300 data_attrs["data-widget-name"] = widget_name
302 return render_to_string(
303 el(
304 "div",
305 el(
306 "div",
307 class_="text-destructive text-lg mb-1",
308 ),
309 el(
310 "p",
311 message,
312 class_="text-sm text-muted-foreground",
313 ),
314 class_="widget-error-card bg-destructive/10 border border-destructive/30 rounded-lg p-4 text-center",
315 **data_attrs,
316 )
317 )
319 @get("/{contributor_id}/health/{check_name}")
320 async def render_health_check(
321 self,
322 request: Request,
323 contributor_id: str,
324 check_name: str,
325 ) -> object:
326 """Render a health check fragment for the HTMX dashboard."""
327 from starlette.responses import HTMLResponse, Response
329 contributor = self._registry.get(contributor_id)
330 if contributor is None:
331 return Response(
332 content=f"Contributor '{contributor_id}' not found",
333 status_code=404,
334 )
336 health_def = next(
337 (h for h in contributor.get_health_definitions() if h.name == check_name),
338 None,
339 )
340 if health_def is not None and not self._has_required_permission(
341 request, health_def.permission
342 ):
343 return Response(content="Permission denied", status_code=403)
345 result = await contributor.render_health_check(check_name)
347 if result.is_ok():
348 return HTMLResponse(render_content(result.unwrap()))
350 error = result.unwrap_err()
351 return Response(content=str(error), status_code=422)
353 @get("/core/widgets/{name}/config")
354 async def widget_config_popup(
355 self,
356 request: Request,
357 name: str,
358 ) -> object:
359 """Render config popup for a widget."""
360 from starlette.responses import HTMLResponse
362 from lexigram.admin.dashboard.widgets import render_widget_config_popup
364 tenant_id = await resolve_tenant_id(request, default="default")
365 user_id = "default"
367 widget_def = None
368 contributor = None
369 for c in self._registry.get_all():
370 for wdef in c.get_dashboard_widgets():
371 if wdef.name == name:
372 widget_def = wdef
373 contributor = c
374 break
375 if widget_def:
376 break
378 if widget_def is None:
379 return HTMLResponse("Widget not found", status_code=404)
381 if not self._user_has_edit_permission(request):
382 await self._audit(
383 request,
384 success=False,
385 event_type=AdminSecurityEventType.PERMISSION_DENIED,
386 reason="permission_denied",
387 route="widget_config_popup",
388 )
389 return HTMLResponse("Permission denied", status_code=403)
391 schema: list[ConfigField] = getattr(
392 contributor, "get_widget_config_schema", lambda _: []
393 )(name)
395 prefs = (
396 await self._settings_service.get_widget_prefs(tenant_id, user_id)
397 if self._settings_service
398 else {}
399 )
400 enabled = "enabled" not in prefs or name in prefs.get("enabled", [])
401 cfg = prefs.get("configs", {}).get(name, {})
403 html = render_widget_config_popup(name, widget_def.title, schema, cfg, enabled)
404 return HTMLResponse(html)
406 @post("/core/widgets/config")
407 async def save_widget_config(self, request: Request) -> object:
408 """Save a single widget's configuration."""
409 from starlette.responses import HTMLResponse
411 if not self._user_has_edit_permission(request):
412 await self._audit(
413 request,
414 success=False,
415 event_type=AdminSecurityEventType.PERMISSION_DENIED,
416 reason="permission_denied",
417 route="save_widget_config",
418 )
419 return HTMLResponse("Permission denied", status_code=403)
421 tenant_id = await resolve_tenant_id(request, default="default")
422 user_id = "default"
424 data = request.scope.get("admin_form_data")
425 if data is None:
426 data = await request.form()
427 widget_name = data.get("widget_name")
428 enabled = "enabled" in data
429 params = {
430 k.removeprefix("param_"): v
431 for k, v in data.items()
432 if k.startswith("param_")
433 }
435 prefs = (
436 await self._settings_service.get_widget_prefs(tenant_id, user_id)
437 if self._settings_service
438 else {}
439 )
440 enabled_list = prefs.get("enabled", [])
441 if enabled and widget_name not in enabled_list:
442 enabled_list.append(widget_name)
443 elif not enabled and widget_name in enabled_list:
444 enabled_list.remove(widget_name)
446 configs = prefs.get("configs", {})
447 if params:
448 configs[widget_name] = params
449 elif widget_name in configs:
450 del configs[widget_name]
452 prefs["enabled"] = enabled_list
453 prefs["configs"] = configs
454 if self._settings_service:
455 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
456 await self._audit(
457 request,
458 widget_name=widget_name or "",
459 kind="widget_config",
460 )
461 return HTMLResponse("", status_code=204)
463 @post("/core/widgets/reorder")
464 async def reorder_widgets(self, request: Request) -> object:
465 """Save widget order after drag-and-drop."""
466 from starlette.responses import HTMLResponse
468 if not self._user_has_edit_permission(request):
469 await self._audit(
470 request,
471 success=False,
472 event_type=AdminSecurityEventType.PERMISSION_DENIED,
473 reason="permission_denied",
474 route="reorder_widgets",
475 )
476 return HTMLResponse("Permission denied", status_code=403)
478 tenant_id = await resolve_tenant_id(request, default="default")
479 user_id = "default"
481 data = await request.json()
482 order_list = data.get("order", [])
483 prefs = (
484 await self._settings_service.get_widget_prefs(tenant_id, user_id)
485 if self._settings_service
486 else {}
487 )
488 prefs["order"] = {name: idx for idx, name in enumerate(order_list)}
489 if self._settings_service:
490 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
491 await self._audit(request, kind="widget_reorder")
492 return HTMLResponse("", status_code=204)
494 @get("/core/widgets/customize")
495 async def customize_all_widgets(self, request: Request) -> object:
496 """Render full dashboard customization panel with all widgets."""
497 from starlette.responses import HTMLResponse
499 if not self._user_has_edit_permission(request):
500 await self._audit(
501 request,
502 success=False,
503 event_type=AdminSecurityEventType.PERMISSION_DENIED,
504 reason="permission_denied",
505 route="customize_all_widgets",
506 )
507 return HTMLResponse("Permission denied", status_code=403)
509 tenant_id = await resolve_tenant_id(request, default="default")
510 user_id = "default"
512 prefs = (
513 await self._settings_service.get_widget_prefs(tenant_id, user_id)
514 if self._settings_service
515 else {}
516 )
517 enabled_list = prefs.get("enabled", [])
518 has_explicit_prefs = "enabled" in prefs
520 grouped: dict[str, tuple[str, list[Any]]] = {}
521 for contributor in self._registry.get_all():
522 widgets = list(contributor.get_dashboard_widgets())
523 if not widgets:
524 continue
525 toggles: list[Any] = []
526 for wdef in widgets:
527 name = wdef.name
528 enabled = name in enabled_list
529 if not has_explicit_prefs:
530 # No saved prefs yet — dashboard shows everything by default,
531 # so the form must render every widget as enabled.
532 enabled = True
533 toggles.append(
534 el(
535 "label",
536 el(
537 "input",
538 type_="checkbox",
539 name=f"enabled_{name}",
540 value="1",
541 checked="checked" if enabled else None,
542 ),
543 el(
544 "span",
545 wdef.title,
546 class_="truncate text-sm font-medium",
547 ),
548 class_=(
549 "flex items-center gap-2 rounded-lg border border-border "
550 "bg-card px-3 py-2 cursor-pointer select-none "
551 "hover:bg-muted/50 transition-colors"
552 ),
553 )
554 )
555 label = getattr(contributor, "display_name", "") or contributor.name
556 grouped[contributor.name] = (label, toggles)
558 sections: list[Any] = []
559 for label, toggles in grouped.values():
560 sections.append(
561 el(
562 "div",
563 el(
564 "h3",
565 label,
566 class_=(
567 "text-xs font-semibold uppercase tracking-wider "
568 "text-muted-foreground"
569 ),
570 ),
571 el("div", class_="mt-1.5 border-t border-border"),
572 el(
573 "div",
574 *toggles,
575 class_="mt-3 grid grid-cols-2 gap-2",
576 ),
577 )
578 )
580 csrf_token = self._get_csrf_token(request)
582 from lexigram.admin.ui.organisms.admin_slide_over import (
583 render_slide_over_fragment,
584 )
586 form = el(
587 "form",
588 el("input", type_="hidden", name="csrf_token", value=csrf_token or ""),
589 *sections,
590 id="widget-customize-form",
591 **{
592 "hx-post": "/admin/core/widgets/customize/save",
593 "hx-swap": "none",
594 "hx-on:htmx:after-request": "if(event.detail.successful){window.location.reload();}",
595 },
596 class_="space-y-6",
597 )
599 return HTMLResponse(
600 render_slide_over_fragment(
601 title="Customize Dashboard",
602 subtitle="Enable/disable widgets and configure their parameters.",
603 content=form,
604 size="xl",
605 footer=[
606 el(
607 "button",
608 "Cancel",
609 type_="button",
610 **{"x-on:click": "open = false"},
611 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",
612 ),
613 el(
614 "button",
615 "Save All Changes",
616 type_="submit",
617 form="widget-customize-form",
618 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",
619 ),
620 ],
621 )
622 )
624 @post("/core/widgets/customize/save")
625 async def save_all_widget_configs(self, request: Request) -> object:
626 """Save all widget configurations from the customize panel."""
627 from starlette.responses import HTMLResponse
629 if not self._user_has_edit_permission(request):
630 await self._audit(
631 request,
632 success=False,
633 event_type=AdminSecurityEventType.PERMISSION_DENIED,
634 reason="permission_denied",
635 route="save_all_widget_configs",
636 )
637 return HTMLResponse("Permission denied", status_code=403)
639 tenant_id = await resolve_tenant_id(request, default="default")
640 user_id = "default"
642 data = request.scope.get("admin_form_data")
643 if data is None:
644 data = await request.form()
646 enabled_list: list[str] = []
647 configs: dict[str, dict[str, str]] = {}
648 all_widget_names: list[str] = []
650 for contributor in self._registry.get_all():
651 for wdef in contributor.get_dashboard_widgets():
652 all_widget_names.append(wdef.name)
654 for key, val in data.items():
655 if key.startswith("enabled_"):
656 wname = key.removeprefix("enabled_")
657 if wname in all_widget_names and val:
658 enabled_list.append(wname)
659 elif key.startswith("param__"):
660 rest = key.removeprefix("param__")
661 wname, pname = rest.split("__", 1)
662 configs.setdefault(wname, {})[pname] = cast("str", val)
664 existing = (
665 await self._settings_service.get_widget_prefs(tenant_id, user_id)
666 if self._settings_service
667 else {}
668 )
669 prefs = {
670 "enabled": enabled_list,
671 "configs": configs,
672 "order": existing.get("order", {}),
673 }
674 if self._settings_service:
675 await self._settings_service.set_widget_prefs(tenant_id, user_id, prefs)
676 await self._audit(
677 request,
678 kind="widget_customize",
679 widget_count=len(all_widget_names),
680 )
681 return HTMLResponse("", status_code=204)
684__all__ = ["WidgetController"]