Взял за основу
https://django.fun/tutorials/django-channels-uvedo...
Подключение к сокету происходит только после авторизации или обновления JWT-токена
Пытаюсь реализовать систему уведомлений. Есть UserNotificationConsumer
class UserNotificationsConsumer(WebsocketConsumer):
def connect(self):
user = self.scope['user']
async_to_sync(self.channel_layer.group_add)(
f"user_{user.pk}", self.channel_name
)
self.accept()
sync_to_async(user.set_online_status())
sync_to_async(self.send_user_notifications())
def disconnect(self, code):
user = self.scope['user']
user.set_online_status()
async_to_sync(self.channel_layer.group_discard)(
f"user_{user.pk}", self.channel_name
)
def send_user_notifications(self):
"""Срабатывает, когда пользователь устанавливает соединение с сокетом"""
user = self.scope['user']
if user.is_online:
notifications_json = get_user_notifications_that_wasnt_sent(user)
if notifications_json:
self.send(text_data=notifications_json)
def set_online_status(self):
"""Устанавливает онлайн статус"""
if self.is_online:
self.is_online = False
else:
self.is_online = True
self.save()
Модели уведомлений:
class BaseNotification(models.Model):
content = models.TextField(default="У вас новое уведомление!")
url = models.URLField()
class UserNotifications(models.Model):
notification = models.ForeignKey(BaseNotification, on_delete=models.CASCADE)
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
viewed_at = models.DateTimeField(default=None, help_text="Когда просмотренно уведомление", null=True)
was_sent = models.BooleanField(default=False)
Создал декоратор, который срабатывает после создания уведомления, рассылая это уведомление всем online-пользователям
def notifications(create_notifications):
"""create_notifcations - функция, которая создаёт уведомления о чем либо"""
def send_notifications(instanse):
notifications_list = create_notifications(instanse)
if notifications_list:
# создаём список пользователей и pk уведомлений
users, notifications_pk = make_users_and_notifications_pk_lists(notifications_list)
# заполняем словарь уведомлениями для конкретного пользователя
notifications_dict = make_users_notifications_dict(notifications_list, users)
# отправляем уведомления онлайн пользователям
send_users_notifications(notifications_dict, notifications_pk)
return send_notifications
def make_users_and_notifications_pk_lists(notifications_list):
"""Создаём список пользователей и pk уведомлений"""
users, notifications_pk = [], []
for note in notifications_list:
if note.user not in users:
users.append(note.user)
notifications_pk.append(note.pk)
return users, notifications_pk
def make_users_notifications_dict(notifications_list, users):
"""Создаём словарь {user: [notifications]} для каждого online-пользователя"""
notifications_dict = {}
for user in users:
if user.is_online:
notifications_dict[user.pk] = [note for note in notifications_list
if user.pk == note.user.pk]
return notifications_dict
def send_users_notifications(notifications_dict, notifications_pk):
"""Отправляем уведомления всем пользователям"""
# получаем главный канал
channel_layer = get_channel_layer()
for key, notes in notifications_dict.items():
user_channel = f"user_{key}"
sync_to_async(channel_layer.group_send)(
user_channel,
{"type": "user.notifications", "text": serializer_notifications(notes)}
)
# ставим для отправленных уведомлений статус отправлено
UserNotifications.objects.filter(pk__in=notifications_pk).update(was_sent=True)
def serializer_notifications(user_notifications):
"""Сериализуем уведомления"""
serialized_user_notifications = json.loads(
json.dumps(
UserNotificationsModelSerializer(
user_notifications,
many=True
).data
)
)
# сериализуем объекты уведомлений
serialized_notifications = {item.notification.pk: BaseNotificationModelSerializer(item.notification).data
for item in user_notifications}
# заменяем pk на сериализованные объекты уведомлений
for item in serialized_user_notifications:
item['notification'] = serialized_notifications.get(item['notification'])
return str(serialized_user_notifications)
