Coverage for src/lexigram/admin/integrations/monitor.py: 0%

1"""Monitor integration — emits metrics and OTel spans for admin operations.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7if TYPE_CHECKING: 

8 from lexigram.contracts.core.di import ( 

9 ContainerRegistrarProtocol, 

10 ContainerResolverProtocol, 

11 ) 

12 

13 

14class _NoOpMonitor: 

15 def increment( 

16 self, name: str, value: int = 1, tags: dict[str, str] | None = None 

17 ) -> None: 

18 pass 

19 

20 def gauge( 

21 self, name: str, value: float, tags: dict[str, str] | None = None 

22 ) -> None: 

23 pass 

24 

25 def histogram( 

26 self, name: str, value: float, tags: dict[str, str] | None = None 

27 ) -> None: 

28 pass 

29 

30 

31class MonitorIntegration: 

32 """Adapter that emits admin metrics and spans via lexigram-monitor. 

33 

34 Gracefully no-ops when ``lexigram-monitor`` is not installed or the 

35 integration is disabled. 

36 """ 

37 

38 def __init__(self, config: Any) -> None: 

39 self._config = config 

40 self._recorder: Any = None 

41 self._enabled = False 

42 

43 def register(self, container: ContainerRegistrarProtocol) -> None: 

44 from lexigram.admin.config import MonitorIntegrationConfig 

45 from lexigram.admin.integrations._optional import is_installed 

46 

47 cfg = self._config 

48 if not isinstance(cfg, MonitorIntegrationConfig): 

49 cfg = MonitorIntegrationConfig() 

50 if not cfg.enabled: 

51 self._recorder = _NoOpMonitor() 

52 return 

53 if not is_installed("lexigram.monitor"): 

54 self._recorder = _NoOpMonitor() 

55 return 

56 self._enabled = True 

57 

58 async def boot(self, container: ContainerResolverProtocol) -> None: 

59 if not self._enabled: 

60 return 

61 try: 

62 from lexigram.contracts.observability.metrics import MetricsRecorderProtocol 

63 

64 self._recorder = await container.resolve(MetricsRecorderProtocol) 

65 except Exception: # noqa: BLE001 

66 self._recorder = _NoOpMonitor() 

67 

68 async def shutdown(self) -> None: 

69 pass 

70 

71 async def health_check(self) -> dict[str, Any]: 

72 return { 

73 "status": "healthy" 

74 if not isinstance(self._recorder, _NoOpMonitor) 

75 else "noop" 

76 } 

77 

78 def increment( 

79 self, name: str, value: int = 1, tags: dict[str, str] | None = None 

80 ) -> None: 

81 if self._recorder: 

82 self._recorder.increment(name, value, tags) 

83 

84 def gauge( 

85 self, name: str, value: float, tags: dict[str, str] | None = None 

86 ) -> None: 

87 if self._recorder: 

88 self._recorder.gauge(name, value, tags) 

89 

90 def histogram( 

91 self, name: str, value: float, tags: dict[str, str] | None = None 

92 ) -> None: 

93 if self._recorder: 

94 self._recorder.histogram(name, value, tags) 

95 

96 

97__all__ = ["MonitorIntegration"]