Coverage for src/lexigram/admin/dashboard/widgets.py: 89%
150 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Admin dashboard widget types and infrastructure.
3Provides WidgetType, WidgetConfig, DashboardConfig and supporting
4store/registry infrastructure for admin dashboard assembly.
5Widget definitions from lexigram-contracts are also re-exported here.
6"""
8from __future__ import annotations
10from dataclasses import dataclass, field
11from datetime import datetime
12from enum import StrEnum
13from typing import Any, Protocol
15from lexigram.admin.dashboard.page_filters import widget_fetch_url
16from lexigram.contracts.admin.types import (
17 DashboardWidgetDefinition,
18 WidgetCategory,
19 WidgetSize,
20)
21from lexigram.ui import el, render_to_string
24def _render_live_widget_script() -> str:
25 """Shared EventSource connection driving all live widgets on the page.
27 One connection per page, not one per widget — browsers cap concurrent
28 HTTP/1.1 connections per origin (~6), the same constraint the staggered
29 hx-trigger delays above already work around. Each live widget's body
30 element carries data-live-resources (comma-separated resource types,
31 or "*" for broadcast-only widgets like activity); on a matching SSE
32 message this re-triggers that widget's existing htmx load (reconcile
33 via the same snapshot endpoint the widget already renders from — no
34 separate patch/diff wire format).
35 """
36 return (
37 "<script>"
38 "(function(){"
39 "if(window.__lexigramLiveWidgets)return;"
40 "window.__lexigramLiveWidgets=true;"
41 "var es=new EventSource('/admin/_sse/widgets');"
42 "es.onmessage=function(ev){"
43 "var data;"
44 "try{data=JSON.parse(ev.data);}catch(e){return;}"
45 "var resourceType=(data.data||{}).resource_type;"
46 "document.querySelectorAll('[data-live-resources]').forEach(function(el){"
47 "var types=el.getAttribute('data-live-resources').split(',');"
48 "if(types.indexOf('*')!==-1||(resourceType&&types.indexOf(resourceType)!==-1)){"
49 "htmx.trigger(el,'live-refresh');"
50 "}"
51 "});"
52 "};"
53 "})();"
54 "</script>"
55 )
58class WidgetType(StrEnum):
59 """Widget types for legacy dashboard builder."""
61 METRIC = "metric"
62 CHART = "chart"
63 TABLE = "table"
64 TEXT = "text"
65 CUSTOM = "custom"
66 STAT_CARD = "stat_card"
67 ACTIVITY = "activity"
68 HEALTH = "health"
71@dataclass
72class WidgetConfig:
73 """Widget configuration."""
75 id: str
76 type: WidgetType
77 title: str
78 config: dict[str, Any] = field(default_factory=dict)
79 position: dict[str, int] = field(
80 default_factory=lambda: {"x": 0, "y": 0, "w": 1, "h": 1},
81 )
83 def to_dict(self) -> dict[str, Any]:
84 """Serialize to dictionary."""
85 return {
86 "id": self.id,
87 "type": self.type.value,
88 "title": self.title,
89 "config": self.config,
90 "position": self.position,
91 }
93 @classmethod
94 def from_dict(cls, data: dict[str, Any]) -> WidgetConfig:
95 """Deserialize from dictionary."""
96 return cls(
97 id=data["id"],
98 type=WidgetType(data["type"]),
99 title=data["title"],
100 config=data.get("config", {}),
101 position=data.get("position", {"x": 0, "y": 0, "w": 1, "h": 1}),
102 )
105@dataclass
106class DashboardConfig:
107 """Dashboard configuration."""
109 id: str
110 name: str
111 widgets: list[WidgetConfig] = field(default_factory=list)
112 layout: dict[str, Any] = field(default_factory=dict)
113 created_at: datetime = field(default_factory=datetime.now)
114 updated_at: datetime = field(default_factory=datetime.now)
116 def to_dict(self) -> dict[str, Any]:
117 """Serialize to dictionary."""
118 return {
119 "id": self.id,
120 "name": self.name,
121 "widgets": [w.to_dict() for w in self.widgets],
122 "layout": self.layout,
123 "created_at": self.created_at.isoformat(),
124 "updated_at": self.updated_at.isoformat(),
125 }
127 @classmethod
128 def from_dict(cls, data: dict[str, Any]) -> DashboardConfig:
129 """Deserialize from dictionary."""
130 return cls(
131 id=data["id"],
132 name=data["name"],
133 widgets=[WidgetConfig.from_dict(w) for w in data.get("widgets", [])],
134 layout=data.get("layout", {}),
135 created_at=datetime.fromisoformat(
136 data.get("created_at", datetime.now().isoformat()),
137 ),
138 updated_at=datetime.fromisoformat(
139 data.get("updated_at", datetime.now().isoformat()),
140 ),
141 )
144class IWidget(Protocol):
145 """Protocol for dashboard widgets."""
147 async def render(self, config: dict[str, Any]) -> str:
148 """Render widget HTML (async)."""
149 ...
151 def get_default_config(self) -> dict[str, Any]:
152 """Get default widget configuration."""
153 ...
156class IDashboardStore(Protocol):
157 """Protocol for dashboard persistence (async)."""
159 async def save(self, dashboard: DashboardConfig) -> bool:
160 """Save dashboard configuration."""
161 ...
163 async def load(self, dashboard_id: str) -> DashboardConfig | None:
164 """Load dashboard configuration."""
165 ...
167 async def list(self) -> list[DashboardConfig]:
168 """List all dashboards."""
169 ...
171 async def delete(self, dashboard_id: str) -> bool:
172 """Delete dashboard."""
173 ...
176class InMemoryDashboardStore:
177 """In-memory dashboard storage implementation."""
179 def __init__(self) -> None:
180 """Initialize store."""
181 self.dashboards: dict[str, DashboardConfig] = {}
183 async def save(self, dashboard: DashboardConfig) -> bool:
184 """Save dashboard."""
185 dashboard.updated_at = datetime.now()
186 self.dashboards[dashboard.id] = dashboard
187 return True
189 async def load(self, dashboard_id: str) -> DashboardConfig | None:
190 """Load dashboard."""
191 return self.dashboards.get(dashboard_id)
193 async def list(self) -> list[DashboardConfig]:
194 """List all dashboards."""
195 return list(self.dashboards.values())
197 async def delete(self, dashboard_id: str) -> bool:
198 """Delete dashboard."""
199 if dashboard_id in self.dashboards:
200 del self.dashboards[dashboard_id]
201 return True
202 return False
205class WidgetRegistry:
206 """Registry for dashboard widget implementations."""
208 def __init__(self) -> None:
209 """Initialize registry."""
210 self._widgets: dict[str, type[IWidget]] = {}
212 def register(self, widget_type: str, widget_class: type[IWidget]) -> None:
213 """Register widget type."""
214 self._widgets[widget_type] = widget_class
216 def get(self, widget_type: str) -> type[IWidget] | None:
217 """Get widget class by type."""
218 return self._widgets.get(widget_type)
220 def list_types(self) -> list[str]:
221 """List registered widget types."""
222 return list(self._widgets.keys())
224 def create_widget(self, widget_type: str) -> IWidget | None:
225 """Create widget instance."""
226 widget_class = self.get(widget_type)
227 if widget_class:
228 return widget_class()
229 return None
231 def render_contributor_widgets(
232 self,
233 contributor_widgets: list[DashboardWidgetDefinition],
234 width: str = "100%",
235 page_filters: dict[str, Any] | None = None,
236 ) -> str:
237 """Render HTML for all contributor-supplied ``DashboardWidgetDefinition`` items.
239 For each widget definition the registry is consulted using the widget's
240 ``name`` as the lookup key. If a matching ``IWidget`` class is found,
241 a rich card is rendered that loads its content via an HTMX ``hx-get``
242 request to the definition's ``render_endpoint``. When no match is
243 found a lightweight ``<div class="widget-placeholder">`` card is
244 rendered instead so contributors can always see their widget slot.
246 Widget sizing from ``DashboardWidgetDefinition.size`` maps to CSS
247 grid column spans (SMALL=1, MEDIUM=2, LARGE=3, FULL=4). If
248 ``refresh_interval_seconds`` is set, the card body auto-refreshes
249 via ``hx-trigger="every <N>ms"``.
251 Args:
252 contributor_widgets: Widget definitions supplied by contributors.
253 width: CSS ``width`` value applied to each card wrapper.
254 page_filters: Optional page-level filter values appended as query
255 parameters to every widget's fetch URL, so widget render
256 endpoints can react to the page's filter state.
258 Returns:
259 Concatenated HTML string for all widget cards.
260 """
261 if not contributor_widgets:
262 return render_to_string(
263 el(
264 "div",
265 el(
266 "div",
267 class_="text-muted-foreground text-lg mb-1",
268 ),
269 el(
270 "p",
271 "No contributor widgets configured.",
272 class_="text-sm text-muted-foreground",
273 ),
274 class_="widget-empty-state bg-muted border border-border rounded-lg p-6 text-center",
275 )
276 )
278 parts: list[str] = []
279 for widget_index, widget_def in enumerate(contributor_widgets):
280 # Map widget size to grid column span
281 size_col_map = {
282 WidgetSize.SMALL: "",
283 WidgetSize.MEDIUM: "lg:col-span-2",
284 WidgetSize.LARGE: "lg:col-span-3",
285 WidgetSize.FULL: "lg:col-span-4",
286 }
287 col_span = size_col_map.get(widget_def.size, "")
289 # Build refresh trigger if interval is set. Live widgets (declared
290 # via live_resource_types) are pushed to via a shared EventSource
291 # instead of polled — see the script emitted after this loop.
292 is_live = bool(widget_def.live_resource_types)
293 refresh_trigger = ""
294 if (
295 not is_live
296 and widget_def.refresh_interval_seconds
297 and widget_def.refresh_interval_seconds > 0
298 ):
299 interval_ms = widget_def.refresh_interval_seconds * 1000
300 refresh_trigger = f"every {interval_ms}ms"
302 # Resolve a matching IWidget class from the registry.
303 # Primary key is the definition name; fall back to category value.
304 lookup_key = getattr(widget_def, "widget_type", widget_def.name)
305 widget_class = self.get(lookup_key) or self.get(widget_def.name)
307 # Build common HTMX trigger: load on page render, plus optional polling.
308 # Initial loads are staggered so the dashboard does not fire every
309 # widget request at once — that would saturate the browser's HTTP/1.1
310 # connection pool (~6 per origin) and starve sidebar navigation
311 # requests for the whole drain.
312 load_trigger = f"load delay:{widget_index * 350}ms"
313 if refresh_trigger:
314 load_trigger = f"{load_trigger}, {refresh_trigger}"
315 if is_live:
316 load_trigger = f"{load_trigger}, live-refresh"
318 # Build the title bar with optional icon and config cog.
319 icon_html = (
320 el("span", widget_def.icon, class_="widget-icon mr-1")
321 if widget_def.icon
322 else None
323 )
324 title_children: list[Any] = []
325 if icon_html is not None:
326 title_children.append(icon_html)
327 title_children.append(widget_def.title)
329 cog = el(
330 "button",
331 "⚙",
332 **{
333 "hx-get": f"/admin/core/widgets/{widget_def.name}/config",
334 "hx-target": "#slide-over-container",
335 "hx-swap": "innerHTML",
336 "hx-push-url": "false",
337 },
338 class_="opacity-0 group-hover:opacity-100 transition-opacity ml-auto text-muted-foreground hover:text-muted-foreground text-sm cursor-pointer",
339 )
340 title_row = el(
341 "div",
342 *title_children,
343 cog,
344 class_="widget-title text-sm font-semibold text-foreground mb-2 flex items-center",
345 )
347 # Contributor label and description
348 subtitle_parts: list[str] = []
349 if widget_def.contributor:
350 subtitle_parts.append(widget_def.contributor)
351 if widget_def.description:
352 subtitle_parts.append(widget_def.description)
354 # Loading skeleton shown while HTMX request is in flight
355 skeleton = el(
356 "div",
357 el("div", class_="h-4 bg-muted rounded w-3/4 mb-2"),
358 el("div", class_="h-4 bg-muted rounded w-1/2 mb-2"),
359 el("div", class_="h-4 bg-muted rounded w-5/6"),
360 class_="animate-pulse py-2",
361 )
363 card_children: list[Any] = [
364 title_row,
365 ]
367 if subtitle_parts:
368 card_children.append(
369 el(
370 "div",
371 " · ".join(subtitle_parts),
372 class_="text-xs text-muted-foreground mb-3",
373 ),
374 )
376 body_kwargs: dict[str, Any] = {
377 "class": "widget-body",
378 "id": f"widget-{widget_def.name}-body",
379 "hx-get": widget_fetch_url(widget_def.render_endpoint, page_filters),
380 "hx-trigger": load_trigger,
381 "hx-swap": "innerHTML",
382 }
383 if is_live:
384 body_kwargs["data-live-resources"] = ",".join(
385 widget_def.live_resource_types
386 )
387 card_children.append(
388 el(
389 "div",
390 skeleton,
391 **body_kwargs,
392 ),
393 )
395 card = el(
396 "div",
397 *card_children,
398 id=f"widget-card-{widget_def.name}",
399 data_widget_name=widget_def.name,
400 class_=f"widget-card bg-card border border-border rounded-lg shadow p-4 group {col_span}".strip(),
401 style=f"width:{width};",
402 )
404 parts.append(render_to_string(card))
406 if any(w.live_resource_types for w in contributor_widgets):
407 parts.append(_render_live_widget_script())
409 return "".join(parts)
412def render_dashboard_widgets(
413 definitions: list[DashboardWidgetDefinition],
414 registry: WidgetRegistry,
415) -> str:
416 """Render all dashboard widget definitions using the provided registry.
418 Convenience wrapper around :meth:`WidgetRegistry.render_contributor_widgets`
419 for callers that already hold a ``WidgetRegistry`` instance and a list of
420 ``DashboardWidgetDefinition`` objects.
422 Args:
423 definitions: Widget definitions to render.
424 registry: Widget registry to look up implementations.
426 Returns:
427 Concatenated HTML string for all widget cards.
428 """
429 return registry.render_contributor_widgets(definitions)
432from lexigram.admin.dashboard.widget_types import ConfigField
435def render_widget_config_popup(
436 widget_name: str,
437 title: str,
438 fields: list[ConfigField],
439 current_values: dict[str, Any],
440 enabled: bool = True,
441) -> str:
442 """Render HTML for a widget config dialog."""
443 rows: list[Any] = [
444 el(
445 "label",
446 el(
447 "input",
448 type_="checkbox",
449 name="enabled",
450 checked="checked" if enabled else None,
451 ),
452 " Show on dashboard",
453 class_="flex items-center gap-2 text-sm mb-4",
454 )
455 ]
457 for f in fields:
458 if isinstance(f, dict):
459 f = ConfigField(**f)
460 value = current_values.get(f.name, f.default)
461 rows.append(
462 el(
463 "div",
464 el("label", f.label, class_="block text-sm font-medium mb-1"),
465 _render_field_input(f, value),
466 class_="mb-3",
467 ),
468 )
470 from lexigram.admin.ui.organisms.admin_slide_over import (
471 render_slide_over_fragment,
472 )
474 form = el(
475 "form",
476 *rows,
477 el("input", type_="hidden", name="widget_name", value=widget_name),
478 id=f"widget-config-form-{widget_name}",
479 **{
480 "hx-post": "/admin/core/widgets/config",
481 "hx-swap": "none",
482 "hx-on:htmx:after-request": "if(event.detail.successful){window.location.reload();}",
483 },
484 class_="space-y-3",
485 )
487 return render_slide_over_fragment(
488 title=f"Configure: {title}",
489 subtitle="Update this widget's settings.",
490 content=form,
491 size="md",
492 footer=[
493 el(
494 "button",
495 "Cancel",
496 type_="button",
497 **{"x-on:click": "open = false"},
498 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",
499 ),
500 el(
501 "button",
502 "Save",
503 type_="submit",
504 form=f"widget-config-form-{widget_name}",
505 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",
506 ),
507 ],
508 )
511def _render_field_input(
512 field: ConfigField, current: Any, widget_name: str | None = None
513) -> str:
514 prefix = f"param__{widget_name}__" if widget_name else "param_"
515 if field.type == "select" and field.options:
516 opts = [
517 el(
518 "option",
519 label,
520 value=str(val),
521 selected="selected" if val == current else None,
522 )
523 for val, label in field.options
524 ]
525 return str(
526 el(
527 "select",
528 *opts,
529 name=f"{prefix}{field.name}",
530 class_="w-full border rounded px-2 py-1 text-sm",
531 )
532 )
533 if field.type == "number":
534 return str(
535 el(
536 "input",
537 type_="number",
538 name=f"{prefix}{field.name}",
539 value=str(current) if current is not None else "",
540 class_="w-full border rounded px-2 py-1 text-sm",
541 )
542 )
543 if field.type == "boolean":
544 return str(
545 el(
546 "input",
547 type_="checkbox",
548 name=f"{prefix}{field.name}",
549 checked="checked" if current else None,
550 )
551 )
552 return str(
553 el(
554 "input",
555 type_="text",
556 name=f"{prefix}{field.name}",
557 value=str(current) if current is not None else "",
558 class_="w-full border rounded px-2 py-1 text-sm",
559 )
560 )
563__all__ = [
564 "ConfigField",
565 "DashboardConfig",
566 "DashboardWidgetDefinition",
567 "IDashboardStore",
568 "IWidget",
569 "InMemoryDashboardStore",
570 "WidgetCategory",
571 "WidgetConfig",
572 "WidgetRegistry",
573 "WidgetSize",
574 "WidgetType",
575 "render_dashboard_widgets",
576 "render_widget_config_popup",
577]