django で Websocket
- 2019/09/07
- 16:06
django で Websocketを試します。
（参考）
http://www.denzow.me/entry/2018/03/25/235952
環境は以下の通り
python ： 3.7.2
django ： 2.2.4
インストール
django で websocket を使用する際は、channels というライブラリを使用します。
プロジェクト/アプリ作成
websocket を試すためのプロジェクトを作成します。
channelsの有効化
プロジェクトディレクトリ下のsettings.pyの以下太字箇所を修正します。
View作成
プロジェクトディレクトリ下の urls.py の以下太字箇所を修正します。
続いてアプリディレクトリ直下のviews.pyの以下の内容を追記します。
続いて、アプリディレクトリ下にtemplatesディレクトリを作成し、websocket_view.htmlファイルを作成します。
サーバー側の実装
アプリディレクトリにconsumers.pyを作成します。
Websocket接続されたら、1秒ごとにメッセージをpublishします。
続いて、アプリディレクトリ下にrouting.pyを作成して以下の内容を記載します。
続いて、プロジェクトディレクトリ下にrouting.pyを作成して以下の内容を記載します。
実行すると、テキストエリアにメッセージが追加されていきます。
（参考）
http://www.denzow.me/entry/2018/03/25/235952
環境は以下の通り
python ： 3.7.2
django ： 2.2.4
インストール
django で websocket を使用する際は、channels というライブラリを使用します。
$ pip install channels channels_redis
プロジェクト/アプリ作成
websocket を試すためのプロジェクトを作成します。
$ django-admin startproject websocket_proj
$ python manage.py startapp websocket_app
channelsの有効化
プロジェクトディレクトリ下のsettings.pyの以下太字箇所を修正します。
INSTALLED_APPS = [
...
'websocket_app',
'channels',
...
ASGI_APPLICATION = 'websocket_proj.routing.application'
]
View作成
プロジェクトディレクトリ下の urls.py の以下太字箇所を修正します。
from django.contrib import admin
from django.urls import path
from django.conf.urls import url
import websocket_app.views as websocket_view
urlpatterns = [
path('admin/', admin.site.urls),
url(r'^websocket/', websocket_view.WebsocketView.as_view())
]
続いてアプリディレクトリ直下のviews.pyの以下の内容を追記します。
from django.views.generic import TemplateView
class WebsocketView(TemplateView):
template_name = "websocket_view.html"
def get(self, request, *args, **kwargs):
context = super(WebsocketView, self).get_context_data(*kwargs)
return render(self.request, self.template_name, context)
続いて、アプリディレクトリ下にtemplatesディレクトリを作成し、websocket_view.htmlファイルを作成します。
<!DOCTYPE html>
<html lang='ja'>
<head>
<meta charset="utf-8">
<script type="text/javascript" src="https://ajax.googleapis.com/ajax/libs/jquery/1.9.1/jquery.min.js"></script>
<title>websocket test</title>
</head>
<body>
<textarea style="width:300px; height:100px;" id="message_area" readonly></textarea>
</body>
<script>
var ws = new WebSocket('ws://' + window.location.host + '/ws/');
ws.onmessage = function(e) {
var data = JSON.parse(e.data);
var msg = data['message'];
$("#message_area").append(msg + '\n');
};
ws.onclose = function(e) {
$("#message_area").append('websocket closed');
}
</script>
</html>
サーバー側の実装
アプリディレクトリにconsumers.pyを作成します。
Websocket接続されたら、1秒ごとにメッセージをpublishします。
from channels.generic.websocket import WebsocketConsumer
import json
import threading
import time
class TestConsumer(WebsocketConsumer):
def connect(self):
self.accept()
self.start_publish()
def disconnect(self, close_code):
self.stop_publish()
def start_publish(self):
self.publishing = True
self.t = threading.Thread(target=self.interval_publish)
self.t.start()
def stop_publish(self):
self.publishing = False
self.t.join()
def interval_publish(self):
while True:
if self.publishing == False:
break
self.send(text_data=json.dumps({
'message': 'test'
}))
time.sleep(1)
続いて、アプリディレクトリ下にrouting.pyを作成して以下の内容を記載します。
from django.urls import path
from . import consumers
websocket_urlpatterns = [
path('ws/', consumers.TestConsumer),
]
続いて、プロジェクトディレクトリ下にrouting.pyを作成して以下の内容を記載します。
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from websocket_app import routing as ws_routing
application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter(
ws_routing.websocket_urlpatterns
)
),
})
実行すると、テキストエリアにメッセージが追加されていきます。
スポンサーサイト
