Coverage for src/lexigram/web/admin/handlers/request_rate.py: 56%

1"""Request rate widget handler for web admin dashboard.""" 

2 

3from __future__ import annotations 

4 

5from lexigram.contracts.admin import ( 

6 MetricsReadbackProtocol, 

7 Stat, 

8 StatContent, 

9 Tone, 

10 WidgetParams, 

11) 

12from lexigram.contracts.admin.errors import AdminError 

13from lexigram.result import Ok, Result 

14 

15_METRIC_COUNTER = "http_requests_total" 

16 

17 

18class RequestRateWidgetHandler: 

19 """Handler for the request_rate widget. 

20 

21 Args: 

22 metrics: optional metrics readback source; when absent or lacking the 

23 readback capability, the widget degrades to "Not measured". 

24 """ 

25 

26 def __init__(self, metrics: MetricsReadbackProtocol | None = None) -> None: 

27 self._metrics = metrics 

28 

29 async def get_data(self, params: WidgetParams) -> Result[StatContent, AdminError]: 

30 """Fetch request rate data. 

31 

32 Args: 

33 params: Widget request parameters (unused for this widget). 

34 

35 Returns: 

36 Result containing StatContent with request metrics. 

37 """ 

38 if not isinstance(self._metrics, MetricsReadbackProtocol): 

39 return Ok( 

40 StatContent( 

41 stats=( 

42 Stat( 

43 label="Requests/sec", 

44 value="Not measured", 

45 tone=Tone.WARNING, 

46 ), 

47 ) 

48 ) 

49 ) 

50 metric = self._metrics.get_metric(_METRIC_COUNTER) 

51 get_value = getattr(metric, "get_value", None) 

52 value = ( 

53 get_value 

54 if callable(get_value) 

55 else getattr(metric, "get_count", lambda: 0.0) 

56 ) 

57 return Ok( 

58 StatContent( 

59 stats=(Stat(label="Requests/sec", value=f"{float(value()):.1f}"),) 

60 ) 

61 ) 

62 

63 

64__all__ = ["RequestRateWidgetHandler"]