Coverage for src/lexigram/notification/admin/pages/inbox.py: 84%

1"""Inbox management page for /admin/notifications.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.admin import PageContent, PaginationContent 

8from lexigram.contracts.admin.widget_content import ( 

9 EmptyContent, 

10 TableCell, 

11 TableContent, 

12) 

13from lexigram.logging import get_logger 

14from lexigram.notification.inbox.service import InboxService 

15 

16logger = get_logger(__name__) 

17 

18DEFAULT_PER_PAGE = 20 

19MAX_PER_PAGE = 200 

20 

21 

22def _query_int(request: Any, name: str, default: int, lo: int, hi: int) -> int: 

23 try: 

24 value = int(request.query_params.get(name, default)) 

25 except (TypeError, ValueError): 

26 value = default 

27 return max(lo, min(value, hi)) 

28 

29 

30class NotificationsInboxPage: 

31 """Full inbox page with persisted notifications. 

32 

33 Args: 

34 inbox_service: Inbox service. When ``None`` the default 

35 in-memory service is used. 

36 """ 

37 

38 def __init__(self, inbox_service: InboxService | None = None) -> None: 

39 self._inbox_service = inbox_service or InboxService() 

40 

41 async def handle(self, request: Any) -> PageContent: 

42 """Handle request and return the inbox page as structured content. 

43 

44 Args: 

45 request: The ASGI request. 

46 

47 Returns: 

48 Structured page content for the inbox. 

49 """ 

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

51 user_id = getattr(user, "id", None) if user is not None else None 

52 

53 try: 

54 raw_per_page = request.query_params.get( 

55 "per_page" 

56 ) or request.query_params.get("limit") 

57 per_page = ( 

58 int(raw_per_page) if raw_per_page is not None else DEFAULT_PER_PAGE 

59 ) 

60 except (TypeError, ValueError): 

61 per_page = DEFAULT_PER_PAGE 

62 per_page = max(1, min(per_page, MAX_PER_PAGE)) 

63 page = _query_int(request, "page", 1, 1, 10**6) 

64 

65 if user_id is None: 

66 messages: list[Any] = [] 

67 else: 

68 try: 

69 messages = await self._inbox_service.get_inbox(user_id) 

70 except Exception as exc: # noqa: BLE001 — non-fatal page render 

71 logger.warning("inbox_page.load_failed", error=str(exc)) 

72 messages = [] 

73 

74 total = len(messages) 

75 page = min(page, max(1, (total + per_page - 1) // per_page)) 

76 offset = (page - 1) * per_page 

77 page_messages = messages[offset : offset + per_page] 

78 

79 if not messages: 

80 return PageContent( 

81 title="Notifications Inbox", 

82 body=EmptyContent( 

83 title="No Notifications", 

84 message="There are no notifications to display yet.", 

85 icon="inbox", 

86 ), 

87 ) 

88 

89 rows = tuple( 

90 ( 

91 TableCell(str(getattr(msg, "title", ""))), 

92 TableCell(str(getattr(msg, "body", ""))), 

93 TableCell(str(getattr(msg, "created_at", ""))), 

94 ) 

95 for msg in page_messages 

96 ) 

97 

98 return PageContent( 

99 title="Notifications Inbox", 

100 body=TableContent( 

101 columns=("Title", "Body", "Created At"), 

102 rows=rows, 

103 ), 

104 pagination=PaginationContent( 

105 page=page, 

106 total=total, 

107 per_page=per_page, 

108 base_url=str(request.url).split("?")[0], 

109 ), 

110 )