Coverage for src/lexigram/notification/admin/contributor.py: 0%

1"""Admin contributor for lexigram-notification — surfaces the persisted 

2user inbox (InboxService) into the Lexigram admin: bell JSON endpoints, 

3an inbox management page, and health. 

4""" 

5 

6from __future__ import annotations 

7 

8from collections.abc import Sequence 

9from typing import TYPE_CHECKING 

10 

11from lexigram.contracts.admin.contributor import BaseAdminContributor 

12from lexigram.contracts.admin.health_payload import HealthCheckPayload 

13from lexigram.contracts.admin.types import ( 

14 AdminHealthDefinition, 

15 AdminRouteSpec, 

16 ManagementPageDefinition, 

17 NavigationContribution, 

18 PageCategory, 

19) 

20from lexigram.contracts.core.health import HealthStatus 

21from lexigram.notification.admin.handlers.inbox import InboxHandlers 

22from lexigram.notification.inbox.service import InboxService 

23from lexigram.result import Err, Ok 

24 

25if TYPE_CHECKING: 

26 from lexigram.contracts.admin.errors import AdminError 

27 from lexigram.contracts.core.di import ContainerResolverProtocol 

28 from lexigram.result import Result 

29 

30_NAV_ITEMS: tuple[NavigationContribution, ...] = ( 

31 NavigationContribution( 

32 label="Notifications", 

33 url="/admin/notifications", 

34 icon="bell", 

35 group="infrastructure", 

36 order=50, 

37 ), 

38) 

39 

40_HEALTH_DEFS: tuple[AdminHealthDefinition, ...] = ( 

41 AdminHealthDefinition( 

42 name="notifications.inbox", 

43 contributor="notifications", 

44 component="Inbox", 

45 check_endpoint="/admin/notifications/health/inbox", 

46 description="Verifies the inbox store is reachable.", 

47 ), 

48) 

49 

50 

51class NotificationAdminContributor(BaseAdminContributor): 

52 """Admin contributor for the lexigram-notification package. 

53 

54 Surfaces the persisted user inbox: a JSON endpoint backend for the 

55 topbar notification bell (list + unread count), mark-read / 

56 mark-all-read round trips, and the /admin/notifications inbox page. 

57 """ 

58 

59 name = "notifications" 

60 display_name = "Notifications" 

61 group = "infrastructure" 

62 icon = "bell" 

63 priority = 50 

64 

65 def __init__(self) -> None: 

66 self._handlers = InboxHandlers() 

67 

68 async def on_admin_boot(self, container: ContainerResolverProtocol) -> None: 

69 """Resolve the inbox service from the container. 

70 

71 Falls back to an in-memory store when the service is not 

72 registered (the bell and page keep working in-process). 

73 

74 Args: 

75 container: The DI container resolver. 

76 """ 

77 service = await container.resolve_optional(InboxService) 

78 self._handlers = InboxHandlers(service=service) 

79 

80 def get_routes(self) -> Sequence[AdminRouteSpec]: 

81 """Return inbox JSON endpoints for the notification bell. 

82 

83 Returns: 

84 Sequence of AdminRouteSpec for inbox endpoints. 

85 """ 

86 return [ 

87 AdminRouteSpec( 

88 path="/admin/notifications/inbox", 

89 method="GET", 

90 handler=self._handlers.get_inbox, 

91 name="inbox.list", 

92 ), 

93 AdminRouteSpec( 

94 path="/admin/notifications/read/{message_id}", 

95 method="POST", 

96 handler=self._handlers.mark_read, 

97 name="inbox.mark_read", 

98 ), 

99 AdminRouteSpec( 

100 path="/admin/notifications/read-all", 

101 method="POST", 

102 handler=self._handlers.mark_all_read, 

103 name="inbox.mark_all_read", 

104 ), 

105 ] 

106 

107 def get_navigation_items(self) -> Sequence[NavigationContribution]: 

108 """Return navigation items. 

109 

110 Returns: 

111 Sequence of NavigationContribution for notifications nav. 

112 """ 

113 return list(_NAV_ITEMS) 

114 

115 def get_health_definitions(self) -> Sequence[AdminHealthDefinition]: 

116 """Return health check definitions. 

117 

118 Returns: 

119 Sequence of AdminHealthDefinition for inbox health checks. 

120 """ 

121 return list(_HEALTH_DEFS) 

122 

123 def get_management_pages(self) -> Sequence[ManagementPageDefinition]: 

124 """Return management page definitions. 

125 

126 Returns: 

127 Sequence of ManagementPageDefinition for the inbox page. 

128 """ 

129 return [ 

130 ManagementPageDefinition( 

131 name="notifications_inbox", 

132 title="Notifications Inbox", 

133 contributor="notifications", 

134 route_path="/notifications", 

135 handler="lexigram.notification.admin.pages.inbox:NotificationsInboxPage", 

136 category=PageCategory.INFRASTRUCTURE, 

137 icon="bell", 

138 description="Persisted in-app notifications for the current user", 

139 order=50, 

140 ), 

141 ] 

142 

143 async def render_health_check( 

144 self, 

145 check_name: str, 

146 ) -> Result[HealthCheckPayload, AdminError]: 

147 """Render the inbox health check fragment. 

148 

149 Args: 

150 check_name: Name of the health check to render. 

151 

152 Returns: 

153 Ok(payload) for the inbox health check. 

154 """ 

155 from typing import cast as type_cast 

156 

157 from lexigram.contracts.admin.errors import AdminError 

158 

159 if check_name != "notifications.inbox": 

160 return type_cast( 

161 "Result[HealthCheckPayload, AdminError]", 

162 Err(AdminError(f"Unknown health check: {check_name}")), 

163 ) 

164 

165 try: 

166 message = await self._handlers.health() 

167 except Exception as exc: # noqa: BLE001 — non-fatal health probe 

168 return Ok( 

169 HealthCheckPayload( 

170 status=HealthStatus.DEGRADED, 

171 component="Inbox", 

172 detail=str(exc), 

173 ) 

174 ) 

175 return Ok( 

176 HealthCheckPayload( 

177 status=HealthStatus.HEALTHY, 

178 component="Inbox", 

179 detail=message, 

180 ) 

181 ) 

182 

183 

184__all__ = ["NotificationAdminContributor"]