Coverage for src/lexigram/admin/controllers/dashboard.py: 93%
137 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Dashboard controller."""
3from __future__ import annotations
5from collections import defaultdict
6import contextvars
7from datetime import datetime
8from typing import Any
10from starlette.requests import Request
11from starlette.responses import HTMLResponse
13from lexigram.admin.controllers.base import AdminController
14from lexigram.admin.dashboard.assembler import DashboardAssembler
15from lexigram.admin.dashboard.page_filters import (
16 applied_from_query,
17 read_page_filters,
18 render_page_filter_form,
19 save_page_filters,
20)
21from lexigram.admin.dashboard.widgets import WidgetRegistry
22from lexigram.admin.engine.renderer import AdminRenderer
23from lexigram.contracts.admin.types import PageFilterField, WidgetSize
24from lexigram.contracts.web import get
25from lexigram.di.decorators import inject
26from lexigram.ui import el
28# Request-scoped in-memory dict, isolated per async context.
29_request_cache_var: contextvars.ContextVar[dict[str, Any] | None] = (
30 contextvars.ContextVar(
31 "admin_request_cache",
32 default=None,
33 )
34)
37def _get_request_cache() -> dict[str, Any]:
38 """Return the request-scoped cache dict for the current async context."""
39 cache = _request_cache_var.get()
40 return cache if cache is not None else {}
43@inject
44class DashboardController(AdminController):
45 """Controller for managing and viewing dashboards."""
47 prefix = ""
49 # Declared page-level filter schema (Filament HasFiltersForm parity).
50 # Subclass to declare filters; the dashboard renders an apply/reset bar
51 # and propagates the values to widget fetch URLs.
52 page_filters: list[PageFilterField] = []
54 def __init__(
55 self,
56 renderer: AdminRenderer,
57 assembler: DashboardAssembler | None = None,
58 widget_registry: WidgetRegistry | None = None,
59 ):
60 super().__init__(renderer)
61 self.assembler = assembler
62 self.widget_registry = widget_registry
63 self._settings_service: Any = None
65 def _size_to_grid_cols(self, size: WidgetSize) -> int:
66 """Map WidgetSize to CSS grid column span."""
67 mapping = {
68 WidgetSize.SMALL: 1,
69 WidgetSize.MEDIUM: 2,
70 WidgetSize.LARGE: 3,
71 WidgetSize.FULL: 4,
72 }
73 return mapping.get(size, 1)
75 @get("/")
76 async def index(self, request: Request) -> HTMLResponse:
77 """Render the main dashboard overview."""
78 from lexigram.admin.ui.organisms.dashboard.widgets import (
79 ActivityFeed,
80 ActivityItem,
81 HealthEntry,
82 Stat,
83 StatCardGrid,
84 SystemHealthWidget,
85 )
86 from lexigram.ui.core.base import raw
88 dashboard_id = request.query_params.get("id", "default")
90 # Page-level filter state: schema defaults → session → query params
91 filter_state: dict[str, Any] = {}
92 if self.page_filters:
93 filter_state = read_page_filters(request, "dashboard", self.page_filters)
94 if applied_from_query(request, self.page_filters):
95 save_page_filters(request, "dashboard", filter_state)
97 breadcrumbs = self.generate_breadcrumbs(
98 ("Home", "/admin/"),
99 current="Dashboard",
100 )
102 # Use assembler widgets when available; fall back to default overview
103 if self.assembler:
104 user = getattr(getattr(request, "state", None), "user", None)
105 contributor_widgets = list(await self.assembler.get_all_widgets(user=user))
106 else:
107 contributor_widgets = []
109 # Load user preferences for widget visibility and ordering
110 widget_prefs = (
111 await self._settings_service.get_widget_prefs("default", "default")
112 if self._settings_service
113 else {}
114 )
115 # Filter and sort widgets. No saved prefs means everything is on.
116 if "enabled" in widget_prefs:
117 enabled_set = set(widget_prefs["enabled"])
118 contributor_widgets = [
119 w for w in contributor_widgets if w.name in enabled_set
120 ]
121 custom_order = widget_prefs.get("order", {})
122 if custom_order:
123 contributor_widgets.sort(key=lambda w: custom_order.get(w.name, w.order))
125 if contributor_widgets and self.widget_registry:
126 # Render HTMX lazy-load widget cards via the registry, annotating
127 # each fetch URL with the current page filter values
128 rendered_html = self.widget_registry.render_contributor_widgets(
129 contributor_widgets,
130 page_filters=filter_state,
131 )
132 widgets_section = el(
133 "div",
134 raw(rendered_html),
135 id="dashboard-grid",
136 class_="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-4 gap-4",
137 )
138 elif contributor_widgets and not self.widget_registry:
139 # Fallback: basic title rendering when no registry is available
140 rendered_widgets: list[Any] = []
141 for w in contributor_widgets:
142 rendered_widgets.append(
143 el(
144 "div",
145 el("h3", w.title, class_="font-semibold"),
146 class_="bg-card rounded-lg p-4 shadow",
147 )
148 )
149 widgets_section = el(
150 "div",
151 *rendered_widgets,
152 class_="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-4",
153 )
154 else:
155 # Provide a beautiful default overview dashboard
156 default_stats = [
157 Stat(
158 label="Resources",
159 value=str(len(self._get_resource_list(request))),
160 icon="layers",
161 color="blue",
162 description="Registered admin resources",
163 ),
164 Stat(
165 label="Active Now",
166 value="—",
167 icon="users",
168 color="green",
169 description="Live sessions",
170 ),
171 Stat(
172 label="Actions Today",
173 value="—",
174 icon="zap",
175 color="primary",
176 description="Admin operations today",
177 ),
178 Stat(
179 label="Errors (24h)",
180 value="—",
181 icon="alert-triangle",
182 color="red",
183 description="Last 24 hours",
184 ),
185 ]
186 default_activity: list[ActivityItem] = []
187 default_health = [
188 HealthEntry(name="Admin API", status="ok"),
189 ]
190 bottom_row = el(
191 "div",
192 el(
193 "div",
194 ActivityFeed(default_activity, title="Recent Activity"),
195 class_="lg:col-span-2",
196 ),
197 SystemHealthWidget(default_health),
198 class_="grid grid-cols-1 lg:grid-cols-3 gap-4",
199 )
200 widgets_section = el(
201 "div",
202 StatCardGrid(default_stats, cols=4),
203 bottom_row,
204 class_="space-y-6",
205 )
207 # SortableJS drag-and-drop controls + widget config helpers
208 dnd_html = raw("""
209<div id="dashboard-dnd-controls" class="mt-4 text-center">
210 <button id="save-layout-btn"
211 class="hidden bg-primary text-primary-foreground px-4 py-2 rounded hover:bg-primary/90 text-sm">
212 💾 Save Layout
213 </button>
214</div>
215<script>
216(function() {
217 var grid = document.getElementById('dashboard-grid');
218 var saveBtn = document.getElementById('save-layout-btn');
219 var sortableInstance = null;
221 function initSortable() {
222 grid = document.getElementById('dashboard-grid');
223 if (!grid || sortableInstance || typeof Sortable === 'undefined') return;
224 sortableInstance = new Sortable(grid, {
225 animation: 150,
226 handle: '.widget-card',
227 onEnd: function() {
228 saveBtn.classList.remove('hidden');
229 }
230 });
231 }
233 initSortable();
234 if (!window.__adminDashboardListeners) {
235 window.__adminDashboardListeners = 1;
236 document.body.addEventListener('htmx:afterSwap', initSortable);
237 document.body.addEventListener('htmx:afterSwap', function(e) {
238 var t = e.detail && e.detail.target;
239 if (t && window.htmx) { try { htmx.process(t); } catch (err) {} }
240 });
241 }
243 if (saveBtn) {
244 saveBtn.addEventListener('click', async function() {
245 if (!sortableInstance) return;
246 var order = Array.from(grid.querySelectorAll('.widget-card')).map(function(card) {
247 return card.dataset.widgetName;
248 });
249 try {
250 var resp = await fetch('/admin/core/widgets/reorder', {
251 method: 'POST',
252 headers: {'Content-Type': 'application/json'},
253 body: JSON.stringify({order: order})
254 });
255 if (resp.ok) {
256 saveBtn.classList.add('hidden');
257 saveBtn.textContent = '✅ Saved!';
258 setTimeout(function() { saveBtn.textContent = '💾 Save Layout'; }, 2000);
259 }
260 } catch(e) {}
261 });
262 }
263})();
264</script>
265""")
267 customize_btn = el(
268 "button",
269 "⚙ Customize Dashboard",
270 **{
271 "hx-get": "/admin/core/widgets/customize",
272 "hx-target": "#slide-over-container",
273 "hx-swap": "innerHTML",
274 "hx-push-url": "false",
275 },
276 class_="text-sm bg-muted hover:bg-muted px-3 py-1.5 rounded border border-border cursor-pointer",
277 )
279 filter_form = (
280 render_page_filter_form(self.page_filters, filter_state, "/admin/")
281 if self.page_filters
282 else None
283 )
285 content = el(
286 "div",
287 el(
288 "div",
289 el(
290 "h2",
291 dashboard_id,
292 class_="text-2xl font-bold text-foreground",
293 ),
294 customize_btn,
295 class_="flex items-center justify-between",
296 ),
297 filter_form,
298 widgets_section,
299 dnd_html,
300 class_="dashboard-view space-y-6",
301 )
303 return await self._render_with_flash(
304 request, content, f"Dashboard: {dashboard_id}", breadcrumbs
305 )
307 async def _render_with_flash(
308 self,
309 request: Request,
310 content: Any,
311 title: str,
312 breadcrumbs: list[dict[str, Any]] | None,
313 ) -> HTMLResponse:
314 """Render the dashboard inside an admin context so flash messages
315 (e.g. the sign-in toast) are consumed by the shell."""
316 from lexigram.admin.state.context import AdminContextManager
318 async with AdminContextManager(request):
319 return await self.render_admin(
320 request,
321 content,
322 title=title,
323 breadcrumbs=breadcrumbs,
324 )
326 def _get_resource_list(self, request: Request) -> list[str]:
327 """Return a list of registered resource names from app state."""
328 try:
329 app = request.app
330 return sorted(getattr(app.state, "admin_resources", {}).keys())
331 except Exception: # noqa: BLE001
332 return []
334 # MetricProtocol aggregation helpers with caching
336 def get_request_cache(self, request: Request) -> dict[str, Any]:
337 """
338 Get request-scoped cache.
340 Returns the context-var-backed cache dict for the current request.
341 The cache is isolated per async context — no shared state between
342 concurrent requests.
343 """
344 return _get_request_cache()
346 async def aggregate_metric(
347 self,
348 request: Request,
349 metric_name: str,
350 compute_func: Any,
351 use_request_cache: bool = True,
352 ) -> Any:
353 """
354 Aggregate metric with automatic caching.
356 Args:
357 request: Current request
358 metric_name: Name of the metric
359 compute_func: Async callable to compute metric value
360 use_request_cache: Use request-scoped cache (default: True)
362 Returns:
363 Computed or cached metric value
365 Example:
366 ```python
367 total_users = await self.aggregate_metric(
368 request,
369 "total_users",
370 lambda: db.execute("SELECT COUNT(*) FROM users")
371 )
372 ```
373 """
374 if use_request_cache:
375 cache = self.get_request_cache(request)
376 key = f"metric:{metric_name}"
377 if key not in cache:
378 cache[key] = await compute_func()
379 return cache[key]
380 return await compute_func()
382 # Time-series helpers
384 def format_time_series(
385 self,
386 data: list[tuple[datetime, float]],
387 interval: str = "hour",
388 ) -> list[dict[str, Any]]:
389 """
390 Format time-series data for charting libraries.
392 Args:
393 data: List of (timestamp, value) tuples
394 interval: Time interval ('hour', 'day', 'week', 'month')
396 Returns:
397 Formatted data for charting (Chart.js compatible)
399 Example:
400 ```python
401 raw_data = [(datetime(2024, 1, 1, 10), 42.0), ...]
402 chart_data = self.format_time_series(raw_data, interval='day')
403 ```
404 """
405 return [
406 {
407 "timestamp": ts.isoformat(),
408 "value": value,
409 "label": self._format_timestamp_label(ts, interval),
410 }
411 for ts, value in data
412 ]
414 def _format_timestamp_label(self, ts: datetime, interval: str) -> str:
415 """Format timestamp for display based on interval."""
416 if interval == "hour":
417 return ts.strftime("%H:%M")
418 if interval == "day":
419 return ts.strftime("%Y-%m-%d")
420 if interval == "week":
421 return f"Week {ts.isocalendar()[1]}, {ts.year}"
422 if interval == "month":
423 return ts.strftime("%b %Y")
424 return ts.isoformat()
426 def aggregate_time_series(
427 self,
428 data: list[tuple[datetime, float]],
429 interval: str = "hour",
430 ) -> list[tuple[datetime, float]]:
431 """
432 Aggregate time-series data into intervals.
434 Groups data points by time interval and sums/averages values.
436 Args:
437 data: List of (timestamp, value) tuples
438 interval: Aggregation interval ('hour', 'day', 'week', 'month')
440 Returns:
441 Aggregated time-series data
442 """
443 buckets: dict[str, list[float]] = defaultdict(list)
445 for ts, value in data:
446 bucket_key = self._get_bucket_key(ts, interval)
447 buckets[bucket_key].append(value)
449 # Compute average for each bucket
450 result: list[tuple[datetime, float]] = []
451 for key, values in sorted(buckets.items()):
452 avg_value = sum(values) / len(values)
453 # Parse bucket key back to datetime
454 bucket_ts = self._parse_bucket_key(key, interval)
455 result.append((bucket_ts, avg_value))
457 return result
459 def _get_bucket_key(self, ts: datetime, interval: str) -> str:
460 """Generate bucket key for timestamp."""
461 if interval == "hour":
462 return ts.strftime("%Y-%m-%d %H:00")
463 if interval == "day":
464 return ts.strftime("%Y-%m-%d")
465 if interval == "week":
466 iso_year, iso_week, _ = ts.isocalendar()
467 return f"{iso_year}-W{iso_week:02d}"
468 if interval == "month":
469 return ts.strftime("%Y-%m")
470 return ts.isoformat()
472 def _parse_bucket_key(self, key: str, interval: str) -> datetime:
473 """Parse bucket key back to datetime."""
474 if interval == "hour":
475 return datetime.strptime(key, "%Y-%m-%d %H:00")
476 if interval == "day":
477 return datetime.strptime(key, "%Y-%m-%d")
478 if interval == "week":
479 year, week = key.split("-W")
480 # ISO week to datetime
481 return datetime.strptime(f"{year} {week} 1", "%Y %W %w")
482 if interval == "month":
483 return datetime.strptime(key, "%Y-%m")
484 return datetime.fromisoformat(key)