-
-
Notifications
You must be signed in to change notification settings - Fork 601
/
webpush.py
55 lines (49 loc) · 1.51 KB
/
webpush.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
import warnings
from pywebpush import WebPushException, webpush
from .conf import get_manager
from .exceptions import WebPushError
def get_subscription_info(application_id, uri, browser, auth, p256dh):
if uri.startswith("https://"):
endpoint = uri
else:
url = get_manager().get_wp_post_url(application_id, browser)
endpoint = "{}/{}".format(url, uri)
warnings.warn(
"registration_id should be the full endpoint returned from pushManager.subscribe",
DeprecationWarning,
stacklevel=2,
)
return {
"endpoint": endpoint,
"keys": {
"auth": auth,
"p256dh": p256dh,
}
}
def webpush_send_message(device, message, **kwargs):
subscription_info = get_subscription_info(
device.application_id, device.registration_id,
device.browser, device.auth, device.p256dh)
try:
results = {"results": [{"original_registration_id": device.registration_id}]}
response = webpush(
subscription_info=subscription_info,
data=message,
vapid_private_key=get_manager().get_wp_private_key(device.application_id),
vapid_claims=get_manager().get_wp_claims(device.application_id).copy(),
**kwargs
)
if response.ok:
results["success"] = 1
else:
results["failure"] = 1
results["results"][0]["error"] = response.content
return results
except WebPushException as e:
if e.response is not None and e.response.status_code in [404, 410]:
results["failure"] = 1
results["results"][0]["error"] = e.message
device.active = False
device.save()
return results
raise WebPushError(e.message)