Coverage for src/lexigram/notification/di/inbox_provider.py: 100%

1"""InboxProvider — DI provider for the user inbox subsystem.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING 

6 

7from lexigram.contracts.core import HealthCheckResult, HealthStatus, ProviderPriority 

8from lexigram.contracts.notification.inbox import InboxStoreProtocol 

9from lexigram.di.provider import Provider 

10from lexigram.logging import get_logger 

11from lexigram.notification.config import InboxConfig 

12 

13if TYPE_CHECKING: 

14 from lexigram.contracts.core.di import ( 

15 ContainerRegistrarProtocol, 

16 ContainerResolverProtocol, 

17 ) 

18 

19logger = get_logger(__name__) 

20 

21 

22class InboxProvider(Provider): 

23 """Register the inbox store and service into the DI container. 

24 

25 Honors :class:`~lexigram.notification.config.InboxConfig.store_backend`: 

26 

27 - ``database`` — :class:`~lexigram.notification.inbox.database.DatabaseInboxStore` 

28 backed by :class:`~lexigram.contracts.data.sql.database.DatabaseProviderProtocol` 

29 (resolved lazily at first use). 

30 - ``memory`` — :class:`~lexigram.notification.inbox.memory.InMemoryInboxStore`. 

31 

32 Also registers :class:`~lexigram.notification.inbox.service.InboxService` 

33 for constructor injection. 

34 

35 Args: 

36 config: Inbox configuration. When ``None`` the default 

37 :class:`InboxConfig` (``store_backend="database"``) is used. 

38 """ 

39 

40 name = "inbox" 

41 priority = ProviderPriority.INFRASTRUCTURE 

42 

43 def __init__(self, config: InboxConfig | None = None) -> None: 

44 super().__init__() 

45 self._config = config or InboxConfig() 

46 self._store: InboxStoreProtocol | None = None 

47 

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

49 """Bind inbox store and service into the container. 

50 

51 Args: 

52 container: DI registrar received from the framework. 

53 """ 

54 from lexigram.notification.inbox.service import InboxService 

55 

56 backend = self._config.store_backend 

57 if backend == "database": 

58 from lexigram.contracts.data.sql.database import ( 

59 DatabaseProviderProtocol, 

60 ) 

61 from lexigram.notification.inbox.database import DatabaseInboxStore 

62 

63 async def _database_store_factory( 

64 resolver: ContainerResolverProtocol, 

65 ) -> DatabaseInboxStore: 

66 db = await resolver.resolve(DatabaseProviderProtocol) 

67 return DatabaseInboxStore(db=db) 

68 

69 container.singleton( 

70 InboxStoreProtocol, 

71 factory=_database_store_factory, 

72 ) 

73 else: 

74 from lexigram.notification.inbox.memory import InMemoryInboxStore 

75 

76 container.singleton(InboxStoreProtocol, InMemoryInboxStore) 

77 

78 container.singleton(InboxService, InboxService) 

79 logger.info("inbox_registered", backend=backend) 

80 

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

82 """Resolve the inbox store for health checking. 

83 

84 Args: 

85 container: DI resolver used to obtain the registered store. 

86 """ 

87 from lexigram.notification.inbox.service import InboxService 

88 

89 self._store = await container.resolve(InboxStoreProtocol) 

90 await container.resolve(InboxService) 

91 logger.info("inbox_booted", backend=self._config.store_backend) 

92 

93 async def shutdown(self) -> None: 

94 """Release store reference on shutdown.""" 

95 self._store = None 

96 

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

98 """Delegate health check to the registered inbox store. 

99 

100 Returns degraded status before :meth:`boot` has been called. 

101 

102 Args: 

103 timeout: Max seconds to wait for the store health probe. 

104 

105 Returns: 

106 :class:`~lexigram.contracts.core.HealthCheckResult`. 

107 """ 

108 if self._store is None: 

109 return HealthCheckResult( 

110 component="inbox", 

111 status=HealthStatus.DEGRADED, 

112 message="inbox store not initialized", 

113 ) 

114 

115 store_result = await self._store.health_check(timeout=timeout) 

116 return HealthCheckResult( 

117 component="inbox", 

118 status=store_result.status, 

119 message=store_result.message, 

120 details=store_result.details, 

121 error=store_result.error, 

122 duration_ms=store_result.duration_ms, 

123 ) 

124 

125 

126__all__ = ["InboxProvider"]