-
Notifications
You must be signed in to change notification settings - Fork 0
/
models.py
213 lines (172 loc) · 6.99 KB
/
models.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
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
from __future__ import unicode_literals
from __future__ import print_function
import base64
from django.db import models
from django.db.models.query import QuerySet
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.utils.translation import ugettext_lazy as _
from django.utils.translation import get_language, activate
from django.utils.encoding import python_2_unicode_compatible
from django.utils.six.moves import cPickle as pickle # pylint: disable-msg=F
from .compat import AUTH_USER_MODEL
from notification import backends
DEFAULT_QUEUE_ALL = False
QUEUE_ALL = getattr(settings, "NOTIFICATION_QUEUE_ALL", DEFAULT_QUEUE_ALL)
NOTIFICATION_BACKENDS = backends.load_backends()
NOTICE_MEDIA, NOTICE_MEDIA_DEFAULTS = backends.load_media_defaults(
backends=NOTIFICATION_BACKENDS
)
class LanguageStoreNotAvailable(Exception):
pass
def create_notice_type(label, display, description, **kwargs):
NoticeType.create(label, display, description, **kwargs)
@python_2_unicode_compatible
class NoticeType(models.Model):
label = models.CharField(_("label"), max_length=40)
display = models.CharField(_("display"), max_length=50)
description = models.CharField(_("description"), max_length=100)
# by default only on for media with sensitivity less than or equal to this number
default = models.IntegerField(_("default"))
def __str__(self):
return self.label
class Meta:
verbose_name = _("notice type")
verbose_name_plural = _("notice types")
@classmethod
def create(cls, label, display, description, default=2, verbosity=1):
"""
Creates a new NoticeType.
This is intended to be used by other apps as a post_syncdb manangement step.
"""
try:
notice_type = cls._default_manager.get(label=label)
updated = False
if display != notice_type.display:
notice_type.display = display
updated = True
if description != notice_type.description:
notice_type.description = description
updated = True
if default != notice_type.default:
notice_type.default = default
updated = True
if updated:
notice_type.save()
if verbosity > 1:
print("Updated %s NoticeType" % label)
except cls.DoesNotExist:
cls(label=label, display=display, description=description, default=default).save()
if verbosity > 1:
print("Created %s NoticeType" % label)
class NoticeSetting(models.Model):
"""
Indicates, for a given user, whether to send notifications
of a given type to a given medium.
"""
user = models.ForeignKey(AUTH_USER_MODEL, verbose_name=_("user"))
notice_type = models.ForeignKey(NoticeType, verbose_name=_("notice type"))
medium = models.CharField(_("medium"), max_length=1, choices=NOTICE_MEDIA)
send = models.BooleanField(_("send"))
class Meta:
verbose_name = _("notice setting")
verbose_name_plural = _("notice settings")
unique_together = ("user", "notice_type", "medium")
@classmethod
def for_user(cls, user, notice_type, medium):
try:
return cls._default_manager.get(user=user, notice_type=notice_type, medium=medium)
except cls.DoesNotExist:
default = (NOTICE_MEDIA_DEFAULTS[medium] <= notice_type.default)
setting = cls(user=user, notice_type=notice_type, medium=medium, send=default)
setting.save()
return setting
class NoticeQueueBatch(models.Model):
"""
A queued notice.
Denormalized data for a notice.
"""
pickled_data = models.TextField()
def get_notification_language(user):
"""
Returns site-specific notification language for this user. Raises
LanguageStoreNotAvailable if this site does not use translated
notifications.
"""
if getattr(settings, "NOTIFICATION_LANGUAGE_MODULE", False):
try:
app_label, model_name = settings.NOTIFICATION_LANGUAGE_MODULE.split(".")
model = models.get_model(app_label, model_name)
# pylint: disable-msg=W0212
language_model = model._default_manager.get(user__id__exact=user.id)
if hasattr(language_model, "language"):
return language_model.language
except (ImportError, ImproperlyConfigured, model.DoesNotExist):
raise LanguageStoreNotAvailable
raise LanguageStoreNotAvailable
def send_now(users, label, extra_context=None, sender=None):
"""
Creates a new notice.
This is intended to be how other apps create new notices.
notification.send(user, "friends_invite_sent", {
"spam": "eggs",
"foo": "bar",
)
"""
sent = False
if extra_context is None:
extra_context = {}
notice_type = NoticeType.objects.get(label=label)
current_language = get_language()
for user in users:
# get user language for user from language store defined in
# NOTIFICATION_LANGUAGE_MODULE setting
try:
language = get_notification_language(user)
except LanguageStoreNotAvailable:
language = None
if language is not None:
# activate the user's language
activate(language)
for backend in NOTIFICATION_BACKENDS.values():
if backend.can_send(user, notice_type):
backend.deliver(user, sender, notice_type, extra_context)
sent = True
# reset environment to original language
activate(current_language)
return sent
def send(*args, **kwargs):
"""
A basic interface around both queue and send_now. This honors a global
flag NOTIFICATION_QUEUE_ALL that helps determine whether all calls should
be queued or not. A per call ``queue`` or ``now`` keyword argument can be
used to always override the default global behavior.
"""
queue_flag = kwargs.pop("queue", False)
now_flag = kwargs.pop("now", False)
assert not (queue_flag and now_flag), "'queue' and 'now' cannot both be True."
if queue_flag:
return queue(*args, **kwargs)
elif now_flag:
return send_now(*args, **kwargs)
else:
if QUEUE_ALL:
return queue(*args, **kwargs)
else:
return send_now(*args, **kwargs)
def queue(users, label, extra_context=None, sender=None):
"""
Queue the notification in NoticeQueueBatch. This allows for large amounts
of user notifications to be deferred to a seperate process running outside
the webserver.
"""
if extra_context is None:
extra_context = {}
if isinstance(users, QuerySet):
users = [row["pk"] for row in users.values("pk")]
else:
users = [user.pk for user in users]
notices = []
for user in users:
notices.append((user, label, extra_context, sender))
NoticeQueueBatch(pickled_data=base64.b64encode(pickle.dumps(notices))).save()