Coverage for app / backend / src / couchers / notifications / expo_api.py: 38%

1"""Mobile push notification API clients""" 

2 

3import logging 

4from typing import Any 

5 

6import requests 

7import sentry_sdk 

8 

9logger = logging.getLogger(__name__) 

10 

11 

12def send_expo_push_notification( 

13 *, 

14 token: str, 

15 title: str, 

16 body: str, 

17 data: dict[str, Any] | None = None, 

18 collapse_key: str | None = None, 

19) -> dict[str, Any]: 

20 """Send a push notification via the Expo Push API""" 

21 message: dict[str, Any] = { 

22 "to": token, 

23 "sound": "default", 

24 "title": title, 

25 "body": body, 

26 "data": data or {}, 

27 "priority": "high", 

28 "channelId": "default", 

29 } 

30 

31 if collapse_key: 

32 message["collapseKey"] = collapse_key 

33 

34 try: 

35 response = requests.post( 

36 "https://exp.host/--/api/v2/push/send", 

37 json=message, 

38 headers={ 

39 "Accept": "application/json", 

40 "Accept-encoding": "gzip, deflate", 

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

42 }, 

43 timeout=10, 

44 ) 

45 response.raise_for_status() 

46 result = response.json() 

47 if not isinstance(result, dict): 

48 raise ValueError(f"Expected a dict, but got {result}") 

49 return result 

50 except requests.exceptions.RequestException as exc: 

51 logger.error("Failed to send Expo push notification: %s", exc) 

52 sentry_sdk.set_tag("context", "expo_push_api") 

53 sentry_sdk.set_tag("token_prefix", token[:16]) 

54 sentry_sdk.set_context( 

55 "expo_push", 

56 { 

57 "title": title, 

58 "body_preview": body[:120], 

59 "data": data, 

60 "collapse_key": collapse_key, 

61 }, 

62 ) 

63 sentry_sdk.capture_exception(exc) 

64 raise 

65 

66 

67def get_expo_push_receipts(ticket_ids: list[str]) -> dict[str, Any]: 

68 """Fetch push receipts from Expo API. 

69 

70 Expo push notifications are a two-phase delivery system: 

71 1. Send notification → get ticket ID (immediate) 

72 2. Check receipt → get actual delivery status (after ~15 minutes) 

73 

74 Args: 

75 ticket_ids: List of ticket IDs to check (max 1000 per request) 

76 

77 Returns: 

78 Dict mapping ticket_id -> receipt data, where receipt data contains: 

79 - status: "ok" or "error" 

80 - For errors: details.error (e.g., "DeviceNotRegistered", "MessageTooBig") 

81 """ 

82 if not ticket_ids: 

83 return {} 

84 

85 try: 

86 response = requests.post( 

87 "https://exp.host/--/api/v2/push/getReceipts", 

88 json={"ids": ticket_ids}, 

89 headers={ 

90 "Accept": "application/json", 

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

92 }, 

93 timeout=10, 

94 ) 

95 response.raise_for_status() 

96 result = response.json().get("data", {}) 

97 if not isinstance(result, dict): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 raise ValueError(f"Expected a dict, but got {result}") 

99 return result 

100 except requests.exceptions.RequestException as exc: 

101 logger.error("Failed to fetch Expo push receipts: %s", exc) 

102 sentry_sdk.set_tag("context", "expo_push_receipts") 

103 sentry_sdk.set_context("expo_receipts", {"ticket_count": len(ticket_ids)}) 

104 sentry_sdk.capture_exception(exc) 

105 raise