【Channels】Djangoアプリでリアルタイム更新処理を実装する

2023-06-02
2023-06-02
Django
コーディング

この記事では、Channelsモジュールを使用し、Djangoアプリでリアルタイム更新処理を実装する方法を解説します。

channelの実装の基本は公式ドキュメントで紹介されているチュートリアルの方法を利用します。

最終的には、フロントからのイベントではなく、Djangoアプリ内部からのイベントにより、対象のフロントのページをリアルタイムで更新できるような実装とします。

◆動作検証環境

・ローカル環境：mac Catalina
・Python：3.8.9
・Django：4.2.1
・channels : 4.0.0
・channels-redeis : 4.1.0
・daphne : 4.0.0・ redis-server : 7.0.1

1 モジュールのインストール
2 Djangoプロジェクトの作成と各ファイルの作成、編集
3 プロジェクトの動作確認
4 Djangoプロジェクト内でのwebsocket送信の実装

スポンサードリンク

モジュールのインストール

channels、daphneのインストール

今回使用するchannelのバージョンは4.0.0となります。公式ドキュメントの方法に沿い以下のモジュールをインストールします。

python -m pip install -U channels["daphne"]

Djangoのインストール

pip install Django==4.2.1

channels_redisのインストール

1
2
3
 
python3 -m pip install channels_redis
 

スポンサードリンク

redisのインストール

公式ドキュメントの方法ではDockerを利用する方法が紹介されていますが、今回はHomebrewのモジュールを利用する方法をとります。

brew install redis

以下のコマンドで起動します。

以下のコマンドの場合、macOS を再起動しても redis が自動的に起動します。

brew services start redis

redisの起動を停止する際は以下のコマンドを実行します。

brew services stop redis

Djangoプロジェクトの作成と各ファイルの作成、編集

公式ドキュメントの方法に沿い、アプリやファイルを作成すると以下のような構成となります。

対応するファイルを編集していきます。

chat_test/settings.py

...

INSTALLED_APPS = [

'daphne',

"django.contrib.admin",

"django.contrib.auth",

"django.contrib.contenttypes",

"django.contrib.sessions",

"django.contrib.messages",

"django.contrib.staticfiles",

'chat',

]

...

ASGI_APPLICATION = "chat_test.asgi.application"

...

CHANNEL_LAYERS = {

"default": {

"BACKEND": "channels_redis.core.RedisChannelLayer",

"CONFIG": {

"hosts": [("127.0.0.1", 6379)],

}

...

chat/templates/chat/index.html

<!DOCTYPE html>

<html>

<head>

<title>Chat Rooms</title>

</head>

<body>

What chat room would you like to enter?<br>

document.querySelector('#room-name-input').focus();

document.querySelector('#room-name-input').onkeyup = function(e) {

if (e.keyCode === 13) { // enter, return

document.querySelector('#room-name-submit').click();

}

};

document.querySelector('#room-name-submit').onclick = function(e) {

var roomName = document.querySelector('#room-name-input').value;

window.location.pathname = '/chat/' + roomName + '/';

};

</script>

</body>

</html>

chat/templates/chat/room.html

<!DOCTYPE html>

<html>

<head>

</head>

<body>

const roomName = JSON.parse(document.getElementById('room-name').textContent);

const chatSocket = new WebSocket(

'ws://'

+ window.location.host

+ '/ws/chat/'

+ roomName

+ '/'

);

chatSocket.onmessage = function(e) {

const data = JSON.parse(e.data);

document.querySelector('#chat-log').value += (data.message + '\n');

};

chatSocket.onclose = function(e) {

console.error('Chat socket closed unexpectedly');

};

document.querySelector('#chat-message-input').focus();

document.querySelector('#chat-message-input').onkeyup = function(e) {

if (e.keyCode === 13) { // enter, return

document.querySelector('#chat-message-submit').click();

}

};

document.querySelector('#chat-message-submit').onclick = function(e) {

const messageInputDom = document.querySelector('#chat-message-input');

const message = messageInputDom.value;

chatSocket.send(JSON.stringify({

'message': message

}));

messageInputDom.value = '';

};

</script>

</body>

</html>

chat_test/urls.py

from django.contrib import admin

from django.urls import include, path

urlpatterns = [

path("chat/", include("chat.urls")),

path("admin/", admin.site.urls),

]

chat_test/asgi.py

import os

from channels.auth import AuthMiddlewareStack

from channels.routing import ProtocolTypeRouter, URLRouter

from channels.security.websocket import AllowedHostsOriginValidator

from django.core.asgi import get_asgi_application

from chat.routing import websocket_urlpatterns

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "chat_test.settings")

application = ProtocolTypeRouter(

{

"http": get_asgi_application(),

"websocket": AllowedHostsOriginValidator(

AuthMiddlewareStack(URLRouter(websocket_urlpatterns))

}

)

chat/views.py

from django.shortcuts import render

def index(request):

return render(request, "chat/index.html")

def room(request, room_name):

return render(request, "chat/room.html", {"room_name": room_name})

chat/urls.py

from django.urls import path

from . import views

urlpatterns = [

path("", views.index, name="index"),

path("<str:room_name>/", views.room, name="room"),

]

chat/consumers.py

import json

from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):

async def connect(self):

self.room_name = self.scope["url_route"]["kwargs"]["room_name"]

self.room_group_name = "chat_%s" % self.room_name

# Join room group

await self.channel_layer.group_add(self.room_group_name, self.channel_name)

await self.accept()

async def disconnect(self, close_code):

# Leave room group

await self.channel_layer.group_discard(self.room_group_name, self.channel_name)

# Receive message from WebSocket

async def receive(self, text_data):

text_data_json = json.loads(text_data)

message = text_data_json["message"]

# Send message to room group

await self.channel_layer.group_send(

self.room_group_name, {"type": "chat_message", "message": message}

)

# Receive message from room group

async def chat_message(self, event):

message = event["message"]

# Send message to WebSocket

await self.send(text_data=json.dumps({"message": message}))

chat/routing.py

from django.urls import re_path

from . import consumers

websocket_urlpatterns = [

re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),

]

プロジェクトの動作確認

DBのマイグレートを行った後、サーバーを起動し、動作を確認します。

python manage.py migrate

python3 manage.py runserver

サーバー起動時のコンソールに以下のような内容が表示され、ASGIとしてDaphneが利用されている事がわかります。

Starting ASGI/Daphne version 4.0.0 development server at http://127.0.0.1:8000/

Djangoプロジェクト内でのwebsocket送信の実装

今回は、Djangoプロジェクト内でのwebsocket送信の動作確認のために、カスタムコマンドを利用して送信を行います。

カスタムコマンド使用のために、必要なディレクトリとファイルを作成します。

chat/management/commands/ws_test.py

from django.core.management.base import BaseCommand

from asgiref.sync import async_to_sync

from channels.layers import get_channel_layer

class Command(BaseCommand):

def handle(self, *args, **options):

channel_layer = get_channel_layer()

async_to_sync(channel_layer.group_send)(

'chat_lobby',

{

'type': 'chat_message',

'message': "event_trigered_from_custom_command"

}

)

上のコードで実行した場合、index.htmlで表示される「What chat room would you like to enter?」のインプット欄はlobby とする必あります。

これは、接続するグループをlobbyとしているためで、他の文字列をグループとして接続した場合、カスタムコマンドからの送信はおこなわれません。

chat room名を指定してwebsocketを接続してから、カスタムコマンドを実行すると、event_trigered_from_custom_command が表示されます。

上記のカスタムコマンドのコードは、consumer.py内のreceive関数とほぼ同じ実装内容となっております。

カスタムコマンドの場合、ChatConsumerクラスのインスタンス変数が使えないので、channel_layerをget_channel_layer()を利用して定義する事、WSのグループを予め決めていたものでハードコーディングしています。

以上、Channelsを用い、Djangoアプリでリアルタイム更新処理を実装する方法を解説しました。