Coverage for src/couchers/notifications/expo_api.py: 39%

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 return response.json() 

47 except requests.exceptions.RequestException as exc: 

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

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

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

51 sentry_sdk.set_context( 

52 "expo_push", 

53 { 

54 "title": title, 

55 "body_preview": body[:120], 

56 "data": data, 

57 "collapse_key": collapse_key, 

58 }, 

59 ) 

60 sentry_sdk.capture_exception(exc) 

61 raise 

62 

63 

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

65 """Fetch push receipts from Expo API. 

66 

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

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

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

70 

71 Args: 

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

73 

74 Returns: 

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

76 - status: "ok" or "error" 

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

78 """ 

79 if not ticket_ids: 

80 return {} 

81 

82 try: 

83 response = requests.post( 

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

85 json={"ids": ticket_ids}, 

86 headers={ 

87 "Accept": "application/json", 

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

89 }, 

90 timeout=10, 

91 ) 

92 response.raise_for_status() 

93 return response.json().get("data", {}) 

94 except requests.exceptions.RequestException as exc: 

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

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

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

98 sentry_sdk.capture_exception(exc) 

99 raise