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

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 ios_subtitle: str | None = None, 

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

19 collapse_key: str | None = None, 

20) -> dict[str, Any]: 

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

22 message: dict[str, Any] = { 

23 "to": token, 

24 "sound": "default", 

25 "title": title, 

26 "body": body, 

27 "data": data or {}, 

28 "priority": "high", 

29 "channelId": "default", 

30 } 

31 if ios_subtitle: 

32 message["subtitle"] = ios_subtitle 

33 

34 if collapse_key: 

35 message["collapseKey"] = collapse_key 

36 

37 try: 

38 response = requests.post( 

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

40 json=message, 

41 headers={ 

42 "Accept": "application/json", 

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

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

45 }, 

46 timeout=10, 

47 ) 

48 response.raise_for_status() 

49 result = response.json() 

50 if not isinstance(result, dict): 

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

52 return result 

53 except requests.exceptions.RequestException as exc: 

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

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

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

57 sentry_sdk.set_context( 

58 "expo_push", 

59 { 

60 "title": title, 

61 "body_preview": body[:120], 

62 "data": data, 

63 "collapse_key": collapse_key, 

64 }, 

65 ) 

66 sentry_sdk.capture_exception(exc) 

67 raise 

68 

69 

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

71 """Fetch push receipts from Expo API. 

72 

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

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

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

76 

77 Args: 

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

79 

80 Returns: 

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

82 - status: "ok" or "error" 

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

84 """ 

85 if not ticket_ids: 

86 return {} 

87 

88 try: 

89 response = requests.post( 

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

91 json={"ids": ticket_ids}, 

92 headers={ 

93 "Accept": "application/json", 

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

95 }, 

96 timeout=10, 

97 ) 

98 response.raise_for_status() 

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

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

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

102 return result 

103 except requests.exceptions.RequestException as exc: 

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

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

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

107 sentry_sdk.capture_exception(exc) 

108 raise