Coverage for src/lexigram/notification/backends/push/fcm.py: 92%

1"""FCM (Firebase Cloud Messaging) push notification backend.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import TYPE_CHECKING, Any 

7import uuid 

8 

9from lexigram.contracts.core import HealthCheckResult, HealthStatus 

10from lexigram.contracts.mailer import MessageDeliveryReceipt 

11from lexigram.contracts.notification.types import PushMessage 

12from lexigram.logging import get_logger 

13from lexigram.notification.constants import DEFAULT_FCM_TIMEOUT, FCM_SEND_URL 

14from lexigram.notification.exceptions import FCMNotificationError 

15from lexigram.result import Err, Ok, Result 

16 

17if TYPE_CHECKING: 

18 from lexigram.contracts.notification.errors import NotificationError 

19 

20logger = get_logger(__name__) 

21 

22 

23class FCMPush: 

24 """Push notification backend that sends via Firebase Cloud Messaging (FCM). 

25 

26 Implements :class:`~lexigram.contracts.notification.protocols.PushChannelProtocol`. 

27 Requires the ``aiohttp`` optional extra. 

28 

29 Args: 

30 server_key: FCM Server API Key. 

31 timeout: HTTP request timeout in seconds. 

32 """ 

33 

34 def __init__( 

35 self, 

36 server_key: str, 

37 timeout: int = DEFAULT_FCM_TIMEOUT, 

38 ) -> None: 

39 self._server_key = server_key 

40 self._timeout = timeout 

41 

42 async def send( 

43 self, message: PushMessage 

44 ) -> Result[MessageDeliveryReceipt, NotificationError]: 

45 """Send a push notification via FCM REST API. 

46 

47 Args: 

48 message: The push notification to deliver. 

49 

50 Returns: 

51 ``Ok(MessageDeliveryReceipt)`` on success. 

52 ``Err(FCMNotificationError)`` for failures. 

53 

54 Raises: 

55 aiohttp.ClientError: For network connectivity issues. 

56 OSError: For low-level socket/infrastructure errors. 

57 """ 

58 import aiohttp 

59 

60 headers = { 

61 "Authorization": f"key={self._server_key}", 

62 "Content-Type": "application/json", 

63 } 

64 

65 # Build FCM notification payload 

66 payload: dict[str, Any] = { 

67 "registration_ids": message.to, 

68 "notification": { 

69 "title": message.title, 

70 "body": message.body, 

71 }, 

72 } 

73 

74 # Add optional fields 

75 if message.data: 

76 payload["data"] = message.data 

77 if message.image: 

78 payload["notification"]["image"] = message.image 

79 if message.sound: 

80 payload["notification"]["sound"] = message.sound 

81 if message.badge is not None: 

82 payload["notification"]["badge"] = message.badge 

83 if message.ttl is not None: 

84 payload["time_to_live"] = message.ttl 

85 

86 async with aiohttp.ClientSession() as session: 

87 async with session.post( 

88 FCM_SEND_URL, 

89 json=payload, 

90 headers=headers, 

91 timeout=aiohttp.ClientTimeout(total=self._timeout), 

92 ) as resp: 

93 body = await resp.json() 

94 

95 if resp.status == 200: 

96 results = body.get("results", []) 

97 if results and "error" not in results[0]: 

98 message_id = results[0].get("message_id") 

99 receipt = MessageDeliveryReceipt( 

100 message_id=str(uuid.uuid4()), 

101 backend="fcm", 

102 channel="push", 

103 provider_reference=message_id, 

104 ) 

105 logger.info( 

106 "fcm_sent", 

107 to=message.to, 

108 fcm_msg_id=message_id, 

109 success=body.get("success", 0), 

110 ) 

111 return Ok(receipt) 

112 

113 # Handle FCM error in results 

114 if results: 

115 fcm_error = results[0].get("error", "Unknown error") 

116 logger.warning( 

117 "fcm_send_failed", 

118 status=resp.status, 

119 error=fcm_error, 

120 failure_count=body.get("failure", 0), 

121 ) 

122 return Err( 

123 FCMNotificationError( 

124 f"FCM error: {fcm_error}", 

125 fcm_error=fcm_error, 

126 ) 

127 ) 

128 

129 # Handle non-200 HTTP status 

130 error_msg = body.get("error", f"HTTP {resp.status}") 

131 logger.warning("fcm_send_failed", status=resp.status, body=body) 

132 return Err(FCMNotificationError(error_msg)) 

133 

134 async def send_batch( 

135 self, messages: list[PushMessage] 

136 ) -> list[Result[MessageDeliveryReceipt, NotificationError]]: 

137 """Send multiple push notifications in parallel. 

138 

139 Args: 

140 messages: List of push notifications to deliver. 

141 

142 Returns: 

143 List of ``Result`` values, one per message, preserving order. 

144 """ 

145 tasks = [self.send(msg) for msg in messages] 

146 return await asyncio.gather(*tasks) 

147 

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

149 """Check FCM API reachability. 

150 

151 Args: 

152 timeout: Max seconds to wait. 

153 

154 Returns: 

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

156 """ 

157 import aiohttp 

158 

159 try: 

160 async with aiohttp.ClientSession() as session: 

161 async with session.head( 

162 FCM_SEND_URL, 

163 headers={"Authorization": f"key={self._server_key}"}, 

164 timeout=aiohttp.ClientTimeout(total=timeout), 

165 ) as resp: 

166 status = ( 

167 HealthStatus.HEALTHY 

168 if resp.status < 500 

169 else HealthStatus.UNHEALTHY 

170 ) 

171 return HealthCheckResult( 

172 component="fcm", 

173 status=status, 

174 details={"http_status": resp.status}, 

175 ) 

176 except OSError as exc: 

177 return HealthCheckResult( 

178 component="fcm", 

179 status=HealthStatus.UNHEALTHY, 

180 details={"error": str(exc)}, 

181 ) 

182 

183 

184__all__ = ["FCMPush"]