Coverage for src / lexigram / admin / di / sub_providers / integrations.py: 70%

1"""DI sub-provider that registers optional integration adapters. 

2 

3Each optional extension (cache, tasks, search, resilience, storage, features, 

4monitor) has a corresponding ``*Integration`` class in 

5``lexigram.admin.integrations.*``. This sub-provider instantiates them during 

6``register()``, skipping silently when the underlying extension is absent. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import TYPE_CHECKING, Any 

12 

13from lexigram.admin.config import AdminIntegrationsConfig 

14from lexigram.contracts.core.health import HealthCheckResult, HealthStatus 

15 

16if TYPE_CHECKING: 

17 from lexigram.contracts.core.di import ( 

18 ContainerRegistrarProtocol, 

19 ContainerResolverProtocol, 

20 ) 

21 

22 

23class AdminIntegrationsSubProvider: 

24 """Registers and manages all optional integration adapters.""" 

25 

26 def __init__(self, config: AdminIntegrationsConfig) -> None: 

27 self._config = config 

28 self._sub_integrations: list[Any] = [] 

29 

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

31 """Instantiate and register each integration adapter.""" 

32 from lexigram.admin.integrations.cache import CacheIntegration 

33 from lexigram.admin.integrations.features import FeaturesIntegration 

34 from lexigram.admin.integrations.monitor import MonitorIntegration 

35 from lexigram.admin.integrations.resilience import ResilienceIntegration 

36 from lexigram.admin.integrations.search import SearchIntegration 

37 from lexigram.admin.integrations.storage import StorageIntegration 

38 from lexigram.admin.integrations.tasks import TasksIntegration 

39 

40 integrations: list[Any] = [ 

41 CacheIntegration(self._config.cache), 

42 TasksIntegration(self._config.tasks), 

43 SearchIntegration(self._config.search), 

44 ResilienceIntegration(self._config.resilience), 

45 StorageIntegration(self._config.storage), 

46 FeaturesIntegration(self._config.features), 

47 MonitorIntegration(self._config.monitor), 

48 ] 

49 for integ in integrations: 

50 integ.register(container) 

51 self._sub_integrations.append(integ) 

52 

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

54 """Boot all registered integrations and populate the integration registry.""" 

55 from lexigram.admin.integrations import register as _register_integration 

56 

57 for integ in self._sub_integrations: 

58 await integ.boot(container) 

59 _register_integration(integ.__class__.__name__, integ) 

60 

61 async def shutdown(self) -> None: 

62 """Shut down all registered integrations in reverse order.""" 

63 for integ in reversed(self._sub_integrations): 

64 await integ.shutdown() 

65 

66 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

67 """Aggregate health from all integrations.""" 

68 worst = HealthStatus.HEALTHY 

69 details: dict[str, Any] = {} 

70 for integ in self._sub_integrations: 

71 try: 

72 check = await integ.health_check() 

73 details[integ.__class__.__name__] = check.get("status", "unknown") 

74 except Exception as exc: # noqa: BLE001 

75 details[integ.__class__.__name__] = "error" 

76 worst = HealthStatus.DEGRADED 

77 return HealthCheckResult( 

78 component="integrations", 

79 status=worst, 

80 details=details, 

81 ) 

82 

83 

84__all__ = ["AdminIntegrationsSubProvider"]