Coverage for polar/notifications/tasks/push.py: 24%
50 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import TypedDict 1a
2from uuid import UUID 1a
4import structlog 1a
5from exponent_server_sdk import ( 1a
6 DeviceNotRegisteredError,
7 PushClient,
8 PushMessage,
9 PushServerError,
10)
12from polar.notification_recipient.service import ( 1a
13 notification_recipient as notification_recipient_service,
14)
15from polar.notifications.service import notifications 1a
16from polar.worker import AsyncSessionMaker, TaskPriority, actor 1a
18log = structlog.get_logger() 1a
21class PushMessageExtra(TypedDict, total=False): 1a
22 notification_id: str 1a
25_push_client = PushClient() 1a
28def send_push_message( 1a
29 token: str, message: str, extra: PushMessageExtra | None = None
30) -> None:
31 """Send a push message to a specific device token."""
32 try:
33 response = _push_client.publish(
34 PushMessage(
35 to=token,
36 body=message,
37 data=extra,
38 title="Polar",
39 sound="default",
40 ttl=60 * 60 * 24,
41 expiration=None,
42 priority="high",
43 badge=1,
44 category="default",
45 display_in_foreground=True,
46 channel_id="default",
47 subtitle="",
48 mutable_content=False,
49 )
50 )
51 except PushServerError as exc:
52 log.error("notifications.push.server_error", error=str(exc))
53 raise
54 except DeviceNotRegisteredError:
55 log.warning("notifications.push.device_not_registered", token=token)
56 raise
57 except Exception as exc:
58 log.error("notifications.push.unknown_error", error=str(exc))
59 raise
61 try:
62 response.validate_response()
63 except Exception as exc:
64 log.error("notifications.push.validation_error", error=str(exc))
65 raise
68@actor(actor_name="notifications.push", priority=TaskPriority.LOW) 1a
69async def notifications_push(notification_id: UUID) -> None: 1a
70 async with AsyncSessionMaker() as session:
71 notif = await notifications.get(session, notification_id)
72 if not notif:
73 log.warning("notifications.push.not_found")
74 return
76 notification_recipients = await notification_recipient_service.list_by_user(
77 session=session,
78 user_id=notif.user_id,
79 expo_push_token=None,
80 platform=None,
81 )
83 if not notification_recipients:
84 log.warning("notifications.push.devices_not_found", user_id=notif.user_id)
85 return
87 for notification_recipient in notification_recipients:
88 if not notification_recipient.expo_push_token:
89 log.warning(
90 "notifications.push.no_push_token",
91 user_id=notification_recipient.user_id,
92 )
93 continue
95 notification_type = notifications.parse_payload(notif)
96 [subject, _] = notification_type.render()
98 try:
99 send_push_message(
100 token=notification_recipient.expo_push_token,
101 message=subject,
102 extra={"notification_id": str(notification_id)},
103 )
104 except Exception as e:
105 log.error(
106 "notifications.push.send_failed",
107 error=str(e),
108 user_id=notification_recipient.user_id,
109 notification_id=notification_id,
110 )
111 return