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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
from django.apps import apps
from django.contrib import admin, messages
from django.utils.translation import ugettext_lazy as _
from .apns import APNSServerError
from .gcm import GCMError
from .webpush import WebPushError
from .models import APNSDevice, GCMDevice, WNSDevice, WebPushDevice
from .settings import PUSH_NOTIFICATIONS_SETTINGS as SETTINGS
User = apps.get_model(*SETTINGS["USER_MODEL"].split("."))
class DeviceAdmin(admin.ModelAdmin):
list_display = ("__str__", "device_id", "user", "active", "date_created")
list_filter = ("active",)
actions = ("send_message", "send_bulk_message", "enable", "disable")
raw_id_fields = ("user",)
if hasattr(User, "USERNAME_FIELD"):
search_fields = ("name", "device_id", "user__%s" % (User.USERNAME_FIELD))
else:
search_fields = ("name", "device_id")
def send_messages(self, request, queryset, bulk=False):
"""
Provides error handling for DeviceAdmin send_message and send_bulk_message methods.
"""
ret = []
errors = []
r = ""
for device in queryset:
try:
if bulk:
r = queryset.send_message("Test bulk notification")
else:
r = device.send_message("Test single notification")
if r:
ret.append(r)
except GCMError as e:
errors.append(str(e))
except APNSServerError as e:
errors.append(e.status)
except WebPushError as e:
errors.append(e.message)
if bulk:
break
# Because NotRegistered and InvalidRegistration do not throw GCMError
# catch them here to display error msg.
if not bulk:
for r in ret:
if "error" in r["results"][0]:
errors.append(r["results"][0]["error"])
else:
try:
errors = [r["error"] for r in ret[0][0]["results"] if "error" in r]
except TypeError:
for entry in ret[0][0]:
errors = errors + [r["error"] for r in entry["results"] if "error" in r]
if errors:
self.message_user(
request, _("Some messages could not be processed: %r" % (", ".join(errors))),
level=messages.ERROR
)
if ret:
if bulk:
# When the queryset exceeds the max_recipients value, the
# send_message method returns a list of dicts, one per chunk
try:
success = ret[0][0]["success"]
except TypeError:
success = 0
for entry in ret[0][0]:
success = success + entry["success"]
if success == 0:
return
elif len(errors) == len(ret):
return
if errors:
msg = _("Some messages were sent: %s" % (ret))
else:
msg = _("All messages were sent: %s" % (ret))
self.message_user(request, msg)
def send_message(self, request, queryset):
self.send_messages(request, queryset)
send_message.short_description = _("Send test message")
def send_bulk_message(self, request, queryset):
self.send_messages(request, queryset, True)
send_bulk_message.short_description = _("Send test message in bulk")
def enable(self, request, queryset):
queryset.update(active=True)
enable.short_description = _("Enable selected devices")
def disable(self, request, queryset):
queryset.update(active=False)
disable.short_description = _("Disable selected devices")
class GCMDeviceAdmin(DeviceAdmin):
list_display = (
"__str__", "device_id", "user", "active", "date_created", "cloud_message_type"
)
list_filter = ("active", "cloud_message_type")
admin.site.register(APNSDevice, DeviceAdmin)
admin.site.register(GCMDevice, GCMDeviceAdmin)
admin.site.register(WNSDevice, DeviceAdmin)
admin.site.register(WebPushDevice, DeviceAdmin)
|
