Coverage for src/lexigram/admin/contributors/core.py: 89%
149 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 14:56 +0800
1"""Core admin contributor — built-in dashboard surfaces."""
3from __future__ import annotations
5import asyncio
6from collections import deque
7from collections.abc import Sequence
8from importlib.metadata import PackageNotFoundError, version
9import os
10import platform
11from typing import TYPE_CHECKING, Any, cast
13from lexigram.admin.realtime import AdminEvent, SubjectAdminEventHub
14from lexigram.contracts.admin import (
15 ChartContent,
16 ChartPoint,
17 EmptyContent,
18 HealthOverviewProtocol,
19 MetricsReadbackProtocol,
20 NamedHealthCheckProtocol,
21 PageContent,
22 SettingsPanelDefinition,
23 Stat,
24 StatContent,
25 TableCell,
26 TableContent,
27 Tone,
28)
29from lexigram.contracts.admin.contributor import BaseAdminContributor
30from lexigram.contracts.admin.errors import AdminError, WidgetNotFoundError
31from lexigram.contracts.admin.health_payload import HealthCheckPayload
32from lexigram.contracts.admin.types import (
33 AdminHealthDefinition,
34 DashboardWidgetDefinition,
35 NavigationContribution,
36 WidgetCategory,
37 WidgetKind,
38 WidgetParams,
39 WidgetSize,
40 WidgetViewModel,
41)
42from lexigram.contracts.core.health import HealthStatus
43from lexigram.contracts.core.result import Result
44from lexigram.logging import get_logger
45from lexigram.result import Err, Ok
47if TYPE_CHECKING:
48 from lexigram.contracts.core.di import ContainerResolverProtocol
50logger = get_logger(__name__)
53def _status_from_value(value: object) -> HealthStatus:
54 """Map a raw status string to a :class:`HealthStatus` member.
56 Args:
57 value: Raw status string (any case) or enum.
59 Returns:
60 The matching ``HealthStatus``, or ``UNKNOWN`` when unrecognized.
61 """
62 status_value = str(value).lower()
63 if status_value in {"healthy", "degraded", "unhealthy"}:
64 return HealthStatus(status_value)
65 return HealthStatus.UNKNOWN
68def _framework_version() -> str:
69 """Return the installed lexigram core package version, or 'unknown'."""
70 try:
71 return version("lexigram")
72 except PackageNotFoundError:
73 return "unknown"
76class _SystemInfoPageHandler:
77 """Read-only diagnostics panel — proves out the ``get_settings_panels()`` path."""
79 def __init__(self, health: object | None) -> None:
80 self._health = health
82 async def handle(self, request: Any) -> PageContent:
83 """Render framework, runtime, and health diagnostics as a table."""
84 health_status = "unknown"
85 if isinstance(self._health, HealthOverviewProtocol):
86 payload, _details = await self._health.run_all()
87 health_status = str(getattr(payload, "value", "unknown"))
89 rows = (
90 (TableCell(text="Framework Version"), TableCell(text=_framework_version())),
91 (
92 TableCell(text="Python Version"),
93 TableCell(text=platform.python_version()),
94 ),
95 (
96 TableCell(text="Environment"),
97 TableCell(text=os.environ.get("ENVIRONMENT", "unknown")),
98 ),
99 (
100 TableCell(text="Log Level"),
101 TableCell(text=os.environ.get("LOG_LEVEL", "INFO")),
102 ),
103 (TableCell(text="Health Status"), TableCell(text=health_status)),
104 )
105 return PageContent(
106 title="System Info",
107 body=TableContent(columns=("Field", "Value"), rows=rows),
108 )
111class CoreAdminContributor(BaseAdminContributor):
112 """Built-in contributor providing core dashboard surfaces.
114 Provides the framework health overview widget, the main dashboard
115 navigation entry, and the system-wide health check surface.
116 """
118 name = "core"
119 display_name = "Core"
120 group = "system"
121 icon = "layout-dashboard"
122 priority = 0
124 def __init__(
125 self,
126 *,
127 health: object | None = None,
128 metrics: object | None = None,
129 hub: SubjectAdminEventHub | None = None,
130 ) -> None:
131 self._container: ContainerResolverProtocol | None = None
132 self._health = health
133 self._metrics = metrics
134 self._hub = hub
135 self._activity_cache: deque[AdminEvent] = deque(maxlen=50)
136 self._background_tasks: set[asyncio.Task[Any]] = set()
137 self._start_activity_tail()
139 async def on_admin_boot(self, container: object) -> None:
140 self._container = container # type: ignore[assignment]
141 if container is None:
142 return
143 typed_container = cast("ContainerResolverProtocol", container)
144 try:
145 if self._health is None:
146 self._health = await typed_container.resolve_optional(
147 HealthOverviewProtocol
148 )
149 if self._metrics is None:
150 self._metrics = await typed_container.resolve_optional(
151 MetricsReadbackProtocol
152 )
153 if self._hub is None:
154 self._hub = await typed_container.resolve_optional(SubjectAdminEventHub)
155 self._start_activity_tail()
156 except Exception as exc: # noqa: BLE001
157 logger.warning("admin.core_sources_unavailable", error=str(exc))
159 async def on_admin_shutdown(self) -> None:
160 """Cancel the background activity tail, then mark the contributor down."""
161 for task in self._background_tasks:
162 task.cancel()
163 self._background_tasks.clear()
165 def _start_activity_tail(self) -> None:
166 """Tail broadcast admin events into a bounded cache for the activity widget.
168 Broadcast-only: ``subscribe()`` with no ``user_id`` sees only
169 events published without ``target_users``, so per-admin
170 notifications never leak into the shared dashboard. Requires a
171 running event loop; without one (sync construction) the tail
172 starts during ``on_admin_boot`` instead.
173 """
174 if self._hub is None:
175 return
176 try:
177 loop = asyncio.get_running_loop()
178 except RuntimeError:
179 return
180 task = loop.create_task(self._drain_activity())
181 self._background_tasks.add(task)
182 task.add_done_callback(self._background_tasks.discard)
184 async def _drain_activity(self) -> None:
185 if self._hub is None:
186 return
187 async for event in self._hub.subscribe():
188 self._activity_cache.append(event)
190 def get_dashboard_widgets(self) -> Sequence[DashboardWidgetDefinition]:
191 """Return core dashboard widgets: health overview, recent activity, and metrics."""
192 return [
193 DashboardWidgetDefinition(
194 name="health",
195 title="Framework Health",
196 contributor="core",
197 render_endpoint="/admin/core/widgets/health",
198 size=WidgetSize.FULL,
199 category=WidgetCategory.HEALTH,
200 view_kind=WidgetKind.EMPTY,
201 refresh_interval_seconds=10,
202 order=0,
203 icon="heart-pulse",
204 description="Aggregated health status of all framework providers.",
205 ),
206 DashboardWidgetDefinition(
207 name="activity",
208 title="Recent Activity",
209 contributor="core",
210 render_endpoint="/admin/core/widgets/activity",
211 size=WidgetSize.LARGE,
212 category=WidgetCategory.ACTIVITY,
213 view_kind=WidgetKind.TABLE,
214 refresh_interval_seconds=15,
215 order=90,
216 icon="activity",
217 description="Recent admin operations and system events.",
218 live_resource_types=("*",),
219 ),
220 DashboardWidgetDefinition(
221 name="chart_metrics",
222 title="Framework Metrics",
223 contributor="core",
224 render_endpoint="/admin/core/widgets/chart_metrics",
225 size=WidgetSize.FULL,
226 category=WidgetCategory.METRICS,
227 view_kind=WidgetKind.CHART,
228 refresh_interval_seconds=30,
229 order=50,
230 icon="bar-chart-3",
231 description="Key framework performance metrics visualized.",
232 ),
233 ]
235 def get_navigation_items(self) -> Sequence[NavigationContribution]:
236 """Return core navigation: Dashboard link."""
237 return [
238 NavigationContribution(
239 label="Dashboard",
240 url="/admin/",
241 icon="layout-dashboard",
242 group="",
243 order=0,
244 ),
245 ]
247 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]:
248 """Return core health definitions."""
249 return [
250 AdminHealthDefinition(
251 name="admin_core",
252 contributor="core",
253 component="Admin Core",
254 check_endpoint="/admin/core/health/admin_core",
255 icon="shield-check",
256 description="Admin panel core services health.",
257 ),
258 ]
260 def get_settings_panels(self) -> Sequence[SettingsPanelDefinition]:
261 """Contribute the read-only System Info diagnostics panel."""
262 return [
263 SettingsPanelDefinition(
264 name="system-info",
265 title="System Info",
266 contributor=self.package_source,
267 route_path="/admin/system/info",
268 handler=_SystemInfoPageHandler(self._health),
269 icon="info",
270 category="System",
271 order=10,
272 )
273 ]
275 async def render_widget(
276 self,
277 widget_name: str,
278 params: WidgetParams,
279 resolver: ContainerResolverProtocol | None = None,
280 ) -> Result[WidgetViewModel, AdminError]:
281 """Render core widgets.
283 Args:
284 widget_name: Name of the widget to render.
285 params: Widget parameters.
287 Returns:
288 Result containing a WidgetViewModel with structured ``content``,
289 or WidgetNotFoundError if the widget is not found.
290 """
292 if widget_name == "health":
293 if not isinstance(self._health, HealthOverviewProtocol):
294 return self._empty("Health overview", "No health monitor registered.")
295 _payload, details = await self._health.run_all()
296 checks = [
297 *details.get("liveness", {}).get("checks", []),
298 *details.get("readiness", {}).get("checks", []),
299 ]
300 healthy = sum(
301 1
302 for c in checks
303 if str(c.get("status")).lower() == HealthStatus.HEALTHY.value
304 )
305 total = max(len(checks), 1)
306 status_value = getattr(_payload, "value", "unknown")
307 return cast(
308 "Result[WidgetViewModel, AdminError]",
309 Ok(
310 WidgetViewModel(
311 content=StatContent(
312 stats=(
313 Stat(label="Health", value=str(status_value)),
314 Stat(label="Checks", value=f"{healthy}/{total}"),
315 )
316 )
317 )
318 ),
319 )
320 if widget_name == "activity":
321 if self._hub is None and resolver is not None:
322 try:
323 self._hub = await resolver.resolve(SubjectAdminEventHub)
324 self._start_activity_tail()
325 except Exception: # noqa: BLE001 — hub is optional
326 self._hub = None
327 rows: list[tuple[TableCell, TableCell, TableCell]] = []
328 for event in list(self._activity_cache):
329 rows.append(
330 (
331 TableCell(text=str(event.event_type)),
332 TableCell(text=str(event.resource_type or "")),
333 TableCell(
334 text=str(
335 event.resource_id
336 if event.resource_id is not None
337 else ""
338 )
339 ),
340 )
341 )
342 if not rows:
343 return cast(
344 "Result[WidgetViewModel, AdminError]",
345 Ok(
346 WidgetViewModel(
347 content=EmptyContent(
348 title="Recent activity",
349 message="No admin activity yet.",
350 )
351 )
352 ),
353 )
354 return cast(
355 "Result[WidgetViewModel, AdminError]",
356 Ok(
357 WidgetViewModel(
358 content=TableContent(
359 columns=("Event", "Resource", "ID"),
360 rows=tuple(rows),
361 )
362 )
363 ),
364 )
365 if widget_name == "chart_metrics":
366 return cast(
367 "Result[WidgetViewModel, AdminError]",
368 Ok(WidgetViewModel(content=self._chart_metrics_content())),
369 )
370 result: Result[WidgetViewModel, AdminError] = cast(
371 "Result[WidgetViewModel, AdminError]",
372 Err(WidgetNotFoundError(self.name, widget_name)),
373 )
374 return result
376 def _chart_metrics_content(self) -> ChartContent:
377 """Return the framework metrics as structured chart content."""
378 if not isinstance(self._metrics, MetricsReadbackProtocol):
379 return ChartContent(
380 points=(
381 ChartPoint(
382 label="Not measured",
383 value=0.0,
384 tone=Tone.WARNING,
385 ),
386 )
387 )
388 points: list[ChartPoint] = []
389 for name, metric in self._metrics.get_all_metrics().items():
390 value = getattr(metric, "get_value", lambda: 0.0)()
391 points.append(
392 ChartPoint(
393 label=name,
394 value=float(value),
395 tone=Tone.SUCCESS if float(value) >= 0 else Tone.DANGER,
396 )
397 )
398 if not points:
399 metric = self._metrics.get_metric("requests_total")
400 if metric is not None:
401 points.append(
402 ChartPoint(
403 label="requests_total",
404 value=float(getattr(metric, "get_value", lambda: 0.0)()),
405 tone=Tone.DEFAULT,
406 )
407 )
408 return ChartContent(points=tuple(points))
410 def _empty(self, title: str, message: str) -> Result[WidgetViewModel, AdminError]:
411 """Return the empty-content placeholder shape.
413 Args:
414 title: Placeholder title.
415 message: Placeholder message.
417 Returns:
418 Ok containing a WidgetViewModel with EmptyContent.
419 """
420 return cast(
421 "Result[WidgetViewModel, AdminError]",
422 Ok(WidgetViewModel(content=EmptyContent(title=title, message=message))),
423 )
425 async def render_health_check(
426 self,
427 check_name: str,
428 ) -> Result[HealthCheckPayload, AdminError]:
429 """Render health check for admin core.
431 Args:
432 check_name: Name of the health check.
434 Returns:
435 Ok(HealthCheckPayload) with the core status, or
436 Err(AdminError) when the check is unknown.
437 """
438 if check_name == "admin_core":
439 if not isinstance(self._health, HealthOverviewProtocol):
440 return Ok(
441 HealthCheckPayload(
442 status=HealthStatus.UNKNOWN,
443 component="Admin Core",
444 detail="No health registry registered.",
445 )
446 )
447 payload, _details = await self._health.run_all()
448 status = _status_from_value(getattr(payload, "value", "unknown"))
449 return Ok(
450 HealthCheckPayload(
451 status=status,
452 component="Admin Core",
453 detail=(
454 "Admin Core operational"
455 if status == HealthStatus.HEALTHY
456 else "Admin Core degraded"
457 ),
458 )
459 )
460 if not isinstance(self._health, NamedHealthCheckProtocol):
461 return Ok(
462 HealthCheckPayload(
463 status=HealthStatus.UNKNOWN,
464 component="Admin Core",
465 detail="No health registry registered.",
466 )
467 )
468 result = await self._health.run_check(check_name)
469 status_value = str(result.get("status", "UNKNOWN"))
470 return Ok(
471 HealthCheckPayload(
472 status=_status_from_value(status_value),
473 component=str(result.get("component", check_name)),
474 detail=str(
475 result.get("error")
476 or f"{result.get('component', check_name)} checked"
477 ),
478 )
479 )
482__all__ = ["CoreAdminContributor"]