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

1"""Active connections 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_GAUGE = "http_requests_in_progress" 

16 

17 

18class ActiveConnectionsWidgetHandler: 

19 """Handler for the active_connections 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 active connections data. 

31 

32 Args: 

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

34 

35 Returns: 

36 Result containing StatContent with connection metrics. 

37 """ 

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

39 return Ok( 

40 StatContent( 

41 stats=( 

42 Stat( 

43 label="Active", 

44 value="Not measured", 

45 tone=Tone.WARNING, 

46 ), 

47 ) 

48 ) 

49 ) 

50 metric = self._metrics.get_metric(_METRIC_GAUGE) 

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=( 

60 Stat( 

61 label="Active", 

62 value=str(int(float(value()))), 

63 tone=Tone.PRIMARY, 

64 ), 

65 ) 

66 ) 

67 ) 

68 

69 

70__all__ = ["ActiveConnectionsWidgetHandler"]