Django SSE 이용해서 알람 기능 구현 (2)

Django SSE 이용해서 알람 기능 구현 (1)

위 글과 이어지는 내용입니다.

 

이제 ASGI/Channels 설정을 했으니 본격적으로 기능을 구현할 차례이다.

앞서 작성했던 alarms 앱으로 들어가보자.

 

나는 아래의 코드를 보고 내가 필요한 부분을 변경해서 사용한 것이다. 커스텀마이징해서 사용할꺼면 아래 깃허브를 참고하는게 더 도움이 될 것이다.

GitHub - fanout/django-eventstream: Server-Sent Events for Django

 

urls.py

from django.urls import path, include
import django_eventstream
from . import views


urlpatterns = [
        path('events/', views.AlarmList.as_view()),
        path('events/<user_id>/', include(django_eventstream.urls), {
            'format-channels': ['user-{user_id}']
        }),
]

AlarmList 클래스를 'events/' 와 연결해준다.

'events/<user_id>/' 로 들어오면 새로운 채널을 만들거나 기존의 채널이 있으면 연결해준다. - send_event 할때 여기로 연결

views.py

from __future__ import unicode_literals
from rest_framework.permissions import IsAuthenticatedOrReadOnly
from rest_framework_simplejwt.authentication import JWTAuthentication
from rest_framework_simplejwt.tokens import AccessToken
from rest_framework.generics import ListAPIView
from django_eventstream.models import Event
from alarms.serializers import EventSerializers


class AlarmList(ListAPIView):
    queryset = Event.objects.all()
    serializer_class = EventSerializers
    authentication_classes = [JWTAuthentication]
    permission_classes = [IsAuthenticatedOrReadOnly]

    def get_user_instance(self, request, *args, **kwargs):
        auth_header = self.request.headers.get('Authorization')
        access_token = auth_header.split(' ')[1]
        decoded = AccessToken(access_token)
        user_id = decoded['user_id']
        return user_id

    def get(self, request, *args, **kwargs):
        user_id = self.get_user_instance(self.request)
        try:
            room = Event.objects.filter(channel='user-{}'.format(user_id)).order_by('-created')[:10]
            self.queryset = room
        except Event.DoesNotExist:
            # 예외처리

        return self.list(request, *args, **kwargs)

get_user_instance를 통해서 JWT토큰을 파싱하여 사용자 ID를 반환한다.

GET 요청으로 사용자 ID를 가져와 채널이름이 user-<user_id> 인 이벤트를 가지고 온다.

serializers.py

import json
from django_eventstream.models import Event
from rest_framework import serializers

class EventSerializers(serializers.ModelSerializer):

    class Meta:
        model = Event
        fields = [
            "channel",
            "type",
            "data",
            "eid",
            "created",
        ]

    data = serializers.SerializerMethodField("get_data")
    created = serializers.SerializerMethodField("get_created")

    def get_data(self, obj):
        try:
            data = json.loads(obj.data)
        except json.JSONDecodeError:
            data = {}
        return data

    def get_created(self, obj):
        date_format = "%Y-%m-%d %H:%M"
        created = obj.created.strftime(date_format)
        return created

데이터를 직렬화하는 부분이다.

send_event

from django_eventstream import send_event

send_message = f'{user.nickname}님이 댓글을 남겼습니다.\n"{content}"\n{post.title}'

send_event('user-{}'.format(post.author_id), 'message', send_message, json_encode=False)

send_event를 호출해서 채널에 메시지를 전달해줄 수 있다.

첫 인수는 전송할 채널, 두번째는 이벤트 유형, 세번째는 이벤트 데이터이다. 공식 문서에는 여기까지만 나와 았다.

https://github.com/fanout/django-eventstream/blob/master/django_eventstream/eventstream.py

위 링크로 들어가보면 send_event(channel, event_type, data, skip_user_ids=None, async_publish=True, json_encode=True) 로 되어 있어서 나는 json으로 인코딩을 안하고 진하도록 설정하였다.

 

후기

공식 문서만을 무작정 믿고 진행하는게 아니라 코드를 하나하나 분해해서 확인해보는게 조금 더 도움이 되었다.

비록 공식 문서가 잘못되어있는 부분이 많아서 헤매긴했지만 그로 인해 코드를 하나하나 해석하는 능력을 키울 수 있었다.

 

 

Reference

GitHub - fanout/django-eventstream: Server-Sent Events for Django

 

django-eventstream