Coverage for src/lexigram/notification/admin/handlers/inbox.py: 68%

1"""Inbox request handlers for the notification admin contributor. 

2 

3Backs the topbar notification bell with persisted JSON endpoints: 

4list + unread count, mark-read, and mark-all-read. The service is 

5resolved lazily from the request container when available, falling 

6back to the default (in-memory) InboxService otherwise. 

7""" 

8 

9from __future__ import annotations 

10 

11from typing import Any 

12 

13from starlette.responses import JSONResponse 

14 

15from lexigram.logging import get_logger 

16from lexigram.notification.inbox.service import InboxService 

17 

18logger = get_logger(__name__) 

19 

20 

21def _user_id(request: Any) -> Any | None: 

22 """Extract the current user ID from a request. 

23 

24 Real ASGI requests carry the user in ``scope["user"]`` (populated by 

25 the admin auth middleware, mirrored into ``scope["state"]["user"]``). 

26 Minimal hosts that pass plain request stand-ins expose ``.user`` as a 

27 plain attribute instead — read both, never the raising 

28 ``request.user`` property. 

29 

30 Args: 

31 request: The ASGI request. 

32 

33 Returns: 

34 The user ID, or ``None`` when unauthenticated. 

35 """ 

36 scope = getattr(request, "scope", None) 

37 if isinstance(scope, dict): 

38 user = scope.get("user") 

39 if user is None: 

40 state = scope.get("state") 

41 user = state.get("user") if isinstance(state, dict) else None 

42 if isinstance(user, dict): 

43 return user.get("id") 

44 return getattr(user, "id", None) if user is not None else None 

45 

46 user = getattr(request, "user", None) 

47 return getattr(user, "id", None) if user is not None else None 

48 

49 

50def _message_to_dict(message: Any) -> dict[str, Any]: 

51 """Serialize an InboxMessage for the JSON API. 

52 

53 Args: 

54 message: An ``InboxMessage`` instance. 

55 

56 Returns: 

57 JSON-safe message dict. 

58 """ 

59 return { 

60 "id": message.id, 

61 "title": message.title, 

62 "message": message.body, 

63 "read": message.read, 

64 "timestamp": message.created_at.isoformat(), 

65 } 

66 

67 

68class InboxHandlers: 

69 """JSON endpoint handlers for the persisted notification inbox. 

70 

71 Args: 

72 service: Inbox service. When ``None``, the service is resolved 

73 lazily per-request from the request container, falling back 

74 to the default in-memory service. 

75 """ 

76 

77 def __init__(self, service: InboxService | None = None) -> None: 

78 self._service = service 

79 

80 async def _resolve_service(self, request: Any) -> InboxService: 

81 """Resolve an InboxService for the request. 

82 

83 Priority: constructor-injected service > request container > 

84 default in-memory service. 

85 

86 Args: 

87 request: The ASGI request. 

88 

89 Returns: 

90 An InboxService instance. 

91 """ 

92 if self._service is not None: 

93 return self._service 

94 container = getattr(getattr(request, "state", None), "container", None) 

95 if container is None: 

96 app_state = getattr(getattr(request, "app", None), "state", None) 

97 container = getattr(app_state, "container", None) 

98 if container is not None: 

99 try: 

100 service: InboxService = await container.resolve(InboxService) 

101 return service 

102 except Exception as exc: # noqa: BLE001 — non-fatal 

103 logger.warning("inbox_handlers.resolve_failed", error=str(exc)) 

104 return InboxService() 

105 

106 async def get_inbox(self, request: Any) -> JSONResponse: 

107 """Return the current user's inbox as JSON. 

108 

109 Args: 

110 request: The ASGI request. 

111 

112 Returns: 

113 JSON with ``unread_count`` and ``notifications``. 

114 """ 

115 user_id = _user_id(request) 

116 if user_id is None: 

117 return JSONResponse({"unread_count": 0, "notifications": []}) 

118 

119 service = await self._resolve_service(request) 

120 limit = 10 

121 try: 

122 limit = max(1, min(int(request.query_params.get("limit", 10)), 50)) 

123 except (TypeError, ValueError): 

124 limit = 10 

125 

126 messages = await service.get_inbox(user_id, unread_only=False) 

127 unread = await service.count_unread(user_id) 

128 

129 return JSONResponse( 

130 { 

131 "unread_count": unread, 

132 "notifications": [_message_to_dict(m) for m in messages[:limit]], 

133 }, 

134 ) 

135 

136 async def mark_read(self, request: Any) -> JSONResponse: 

137 """Mark a single message as read for the current user. 

138 

139 Args: 

140 request: The ASGI request. 

141 

142 Returns: 

143 JSON acknowledgement. 

144 """ 

145 user_id = _user_id(request) 

146 if user_id is None: 

147 return JSONResponse({"ok": False}, status_code=401) 

148 

149 message_id = request.path_params.get("message_id", "") 

150 service = await self._resolve_service(request) 

151 await service.mark_read(message_id, user_id) 

152 return JSONResponse({"ok": True}) 

153 

154 async def mark_all_read(self, request: Any) -> JSONResponse: 

155 """Mark all of the current user's messages as read. 

156 

157 Args: 

158 request: The ASGI request. 

159 

160 Returns: 

161 JSON acknowledgement. 

162 """ 

163 user_id = _user_id(request) 

164 if user_id is None: 

165 return JSONResponse({"ok": False}, status_code=401) 

166 

167 service = await self._resolve_service(request) 

168 await service.mark_all_read(user_id) 

169 return JSONResponse({"ok": True}) 

170 

171 async def health(self) -> str: 

172 """Probe the underlying inbox store health. 

173 

174 Returns: 

175 Human-readable health status message. 

176 """ 

177 if self._service is None: 

178 return "inbox service not initialized" 

179 result = await self._service._store.health_check() # noqa: SLF001 

180 return f"{result.status.value}: {result.message}" 

181 

182 

183__all__ = ["InboxHandlers"]