Coverage for src/lexigram/admin/ui/observability.py: 31%
94 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-24 23:31 +0800
1"""
2Observability utilities for Lexigram Admin UI.
4Provides metrics collection, logging helpers, and debug tools
5for monitoring HTMX-powered components.
6"""
8from __future__ import annotations
10from functools import wraps
11from time import time
12from typing import TYPE_CHECKING, Any
14from lexigram.logging import get_logger
15from lexigram.serialization import dumps_str
16from lexigram.ui import MetricsCollector, Zones, el, render_to_string
18if TYPE_CHECKING:
19 from collections.abc import Callable
21logger = get_logger(__name__)
24def track_htmx_request(
25 resource: str,
26 target: str,
27 action: str,
28 context: Any | None = None,
29) -> None:
30 """Track an HTMX request."""
31 from lexigram.admin.lib.di import get_admin_resolver
33 resolver = get_admin_resolver(context)
34 metrics = resolver.resolve_sync(MetricsCollector) # type: ignore[attr-defined]
35 metrics.inc(
36 "htmx_requests_total",
37 labels={"resource": resource, "target": target, "action": action},
38 )
41def track_render_time(
42 resource: str,
43 zone: str,
44 duration_ms: float,
45 context: Any | None = None,
46) -> None:
47 """Track component render time."""
48 from lexigram.admin.lib.di import get_admin_resolver
50 resolver = get_admin_resolver(context)
51 metrics = resolver.resolve_sync(MetricsCollector) # type: ignore[attr-defined]
52 metrics.observe(
53 "htmx_render_seconds",
54 duration_ms / 1000,
55 labels={"resource": resource, "zone": zone},
56 )
59def track_error(
60 resource: str,
61 error_type: str,
62 status_code: int,
63 context: Any | None = None,
64) -> None:
65 """Track an error."""
66 from lexigram.admin.lib.di import get_admin_resolver
68 resolver = get_admin_resolver(context)
69 metrics = resolver.resolve_sync(MetricsCollector) # type: ignore[attr-defined]
70 metrics.inc(
71 "htmx_errors_total",
72 labels={
73 "resource": resource,
74 "error_type": error_type,
75 "status_code": str(status_code),
76 },
77 )
80# === Logging Helpers ===
83def log_htmx_request(
84 request: Any,
85 resource: str,
86 state: Any = None,
87) -> None:
88 """Log an HTMX request with structured data."""
89 is_htmx = getattr(request, "headers", {}).get("HX-Request") == "true"
90 target = getattr(request, "headers", {}).get("HX-Target", "")
91 trigger = getattr(request, "headers", {}).get("HX-Trigger", "")
93 logger.info(
94 "htmx_request",
95 resource=resource,
96 is_htmx=is_htmx,
97 target=target,
98 trigger=trigger,
99 method=getattr(request, "method", ""),
100 path=str(getattr(request, "url", "")),
101 state=state.to_query_params()
102 if state and hasattr(state, "to_query_params")
103 else None,
104 )
107def log_htmx_response(
108 resource: str,
109 zone: str,
110 render_time_ms: float,
111 status_code: int = 200,
112) -> None:
113 """Log an HTMX response with timing."""
114 logger.info(
115 "htmx_response",
116 resource=resource,
117 zone=zone,
118 render_time_ms=render_time_ms,
119 status_code=status_code,
120 )
123# === Debug Panel ===
126def render_debug_panel(
127 state: Any = None,
128 zones_info: dict[str, bool] | None = None,
129 render_time_ms: float | None = None,
130) -> str:
131 """
132 Render a debug panel showing current state and zones.
134 Only shown in development mode.
136 Args:
137 state: Current TableState
138 zones_info: Dict of zone_id -> is_rendered
139 render_time_ms: Render time for this request
141 Returns:
142 HTML string for debug panel
143 """
144 from lexigram.logging.debug import is_debug_mode
146 # Only show in debug mode
147 if not is_debug_mode():
148 return ""
150 state_json = "{}"
151 if state and hasattr(state, "to_query_params"):
152 state_json = dumps_str(state.to_query_params(), indent=2)
154 zone_els: list[Any] = []
155 for zone in Zones.all_zones():
156 rendered = zones_info.get(zone.id, False) if zones_info else False
157 status_class = "text-success" if rendered else "text-muted-foreground"
158 status_icon = "●" if rendered else "○"
159 zone_els.append(el("div", f"{status_icon} {zone.id}", class_=status_class))
161 timing_el: Any = ""
162 if render_time_ms is not None:
163 color = (
164 "text-success"
165 if render_time_ms < 50
166 else "text-warning"
167 if render_time_ms < 100
168 else "text-destructive"
169 )
170 timing_el = el("div", f"{render_time_ms:.2f}ms", class_=f"{color} font-mono")
172 return str(
173 render_to_string(
174 el(
175 "div",
176 el(
177 "details",
178 el("summary", "🐛 Debug", class_="cursor-pointer font-medium"),
179 el(
180 "div",
181 el("h4", "State", class_="font-medium mt-2"),
182 el(
183 "pre",
184 state_json,
185 class_="text-xs bg-muted dark:bg-card p-2 rounded overflow-auto max-h-40",
186 ),
187 el("h4", "Zones", class_="font-medium mt-2"),
188 el(
189 "div",
190 *zone_els,
191 class_="text-xs font-mono grid grid-cols-2 gap-1",
192 ),
193 timing_el,
194 class_="mt-2",
195 ),
196 class_="p-2",
197 ),
198 class_="fixed bottom-4 right-4 z-50 bg-card dark:bg-background border border-border rounded-lg shadow-lg text-sm max-w-xs",
199 id="debug-panel",
200 ),
201 ),
202 )
205# === Middleware Decorator ===
208def observe_htmx(resource: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
209 """
210 Decorator to observe HTMX endpoint with metrics and logging.
212 Usage:
213 @observe_htmx("users")
214 async def list_users(request):
215 ...
216 """
218 def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
219 @wraps(func)
220 async def async_wrapper(request: Any, *args: Any, **kwargs: Any) -> Any:
221 start = time()
223 # Log request
224 log_htmx_request(request, resource)
226 # Track request
227 target = getattr(request, "headers", {}).get("HX-Target", "unknown")
228 track_htmx_request(resource, target, "list", context=request)
230 try:
231 result = await func(request, *args, **kwargs)
233 # Log and track response
234 elapsed_ms = (time() - start) * 1000
235 log_htmx_response(resource, target, elapsed_ms)
236 track_render_time(resource, target, elapsed_ms, context=request)
238 return result
239 except Exception as e: # noqa: BLE001 — observability wrapper must capture any exception for error tracking before re-raising
240 # Track error
241 track_error(resource, type(e).__name__, 500, context=request)
242 raise
244 @wraps(func)
245 def sync_wrapper(request: Any, *args: Any, **kwargs: Any) -> Any:
246 start = time()
248 log_htmx_request(request, resource)
249 target = getattr(request, "headers", {}).get("HX-Target", "unknown")
250 track_htmx_request(resource, target, "list", context=request)
252 try:
253 result = func(request, *args, **kwargs)
254 elapsed_ms = (time() - start) * 1000
255 log_htmx_response(resource, target, elapsed_ms)
256 track_render_time(resource, target, elapsed_ms, context=request)
257 return result
258 except Exception as e: # noqa: BLE001 — observability wrapper must capture any exception for error tracking before re-raising
259 track_error(resource, type(e).__name__, 500, context=request)
260 raise
262 # Return appropriate wrapper based on function type
263 import asyncio
265 if asyncio.iscoroutinefunction(func):
266 return async_wrapper
267 return sync_wrapper
269 return decorator
272# === Health Check ===
275def get_health_status(context: Any | None = None) -> dict[str, Any]:
276 """
277 Get health status for the UI system.
279 Returns:
280 Dictionary with health check data
281 """
282 from lexigram.admin.lib.di import get_admin_resolver
284 resolver = get_admin_resolver(context)
285 metrics = resolver.resolve_sync(MetricsCollector) # type: ignore[attr-defined]
286 stats = metrics.to_dict()
288 # Calculate error rate
289 total_requests = sum(
290 v for k, v in stats["counters"].items() if k.startswith("htmx_requests_total")
291 )
292 total_errors = sum(
293 v for k, v in stats["counters"].items() if k.startswith("htmx_errors_total")
294 )
295 error_rate = total_errors / total_requests if total_requests > 0 else 0
297 return {
298 "status": "healthy" if error_rate < 0.05 else "degraded",
299 "total_requests": total_requests,
300 "total_errors": total_errors,
301 "error_rate": error_rate,
302 "metrics": stats,
303 }