DjangoでServer-Sent Eventsを使いwebアプリの画面を定期的に自動更新する

PythonのDjangoでServer-Sent Events(SSE)を使い表示画面を自動で更新する方法をまとめました。

SSEはWebブラウザとサーバーの間で、サーバーからの一方向のリアルタイムイベントストリームを確立するための技術です。

SSEを使用することで、リアルタイムな情報の更新、ユーザーへのプッシュ通知、ストリーミングデータの受信など、さまざまな用途に活用できます。

Djangoアプリの事前準備
実装方法
- 実装に必要なコード
  - イベントを受け取る: JavaScript
  - イベントを送信する: Python
- 受け取ったデータを画面に表示する
終わりに

Djangoアプリの事前準備

Djangoのアプリを作成します。

基本的なdjangoのアプリケーションは、以下のリンクなどを参考に作成可能です。

Writing your first Django app, part 1 | Django documentation | Django

以下の説明は次のディレクトリをもとに説明していきます。

※チュートリアルに従い以下のようにプロジェクトを作成していきました。

以下のコマンドを実行

django-admin startproject eventstreamsite
python manage.py startapp eventstream

eventstreamsite/urls.pyのurlpatternsに次のコードを追加(pathの名称はstartappコマンドで作成したものに置き換えてください。)

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

eventstreamsite/settings.pyのINSTALLED_APPSに次を追加('eventstream'と'Eventstream'の部分はstartappコマンドで作成したものに置き換えてください。)

"eventstream.apps.EventstreamConfig",

実装方法

実装にはイベントを受け取るフロントエンドのJavaScriptとイベントを送信するサーバー側のコードが必要になります。

今回はDjangoのwebアプリでの実装なので、pythonを使いイベントを送信するコードを記載します。

実装に必要なコード

イベントを受け取る: JavaScript

const evtSource = new EventSource("event-stream");
evtSource.onmessage = function(event){
     //イベントを受信した際の処理
     console.log(event.data); //ログにメッセージを出力
}

'event-stream'はイベントを生成するURLを指定します。

メッセージを受け取るとメッセージをログに出力します。

イベントを送信する: Python

eventstream/urls.pyにevent-streamのパスを作成します。

path('event-stream/', views.event_stream, name='event-stream'),

送信するメッセージを作成し、text/event-streamのMIMEタイプで送信を行います。

from django.http import StreamingHttpResponse

def event_stream(request):
     #イベントストリームの送信を開始する前に、必要なレスポンスヘッダーを設定する
     response = StreamingHttpResponse(streaming_content = stream_events())
     response['Content-Type'] = 'text/event-stream'
     response['Cache-control'] = 'no-cache'
     return response

def stream_events():
    #イベントデータを生成して返すジェネレーター関数
    #イベントデータは'data'フィールドにラベル付けされる
    start_time = datetime.datetime.now()
    while True:
        now = datetime.datetime.now()
        elapsed_time = now - start_time
        if elapsed_time.seconds >= 5:
            yield f'data: {now}\n\n'
            start_time = now
        else:
            time.sleep(1)  # 1秒待機

stream_eventsでメッセージを作成し、5秒経過するごとにイベントを送信します。

time.sleep(1)による待機時間はサーバーに負荷をかける可能性があるようなので、実際の環境に合わせて適切に調整してください。

受け取ったデータを画面に表示する

上記のコードにViewなどを追加し、時刻を表示するシンプルな画面を作成します。

eventstream/views.py

from django.shortcuts import render
from django.http import StreamingHttpResponse
import datetime
import time

# IndexのViewを作成
def index(request):
    return render(request, 'eventstream/index.html')

def event_stream(request):
    #イベントストリームの送信を開始する前に、必要なレスポンスヘッダーを設定する
    response = StreamingHttpResponse(streaming_content = stream_events())
    response['Content-Type'] = 'text/event-stream'
    response['Cache-control'] = 'no-cache'
    return response

def stream_events():
    #イベントデータを生成して返すジェネレーター関数
    #イベントデータは'data'フィールドにラベル付けされる
    start_time = datetime.datetime.now()
    while True:
        now = datetime.datetime.now()
        elapsed_time = now - start_time
        if elapsed_time.seconds >= 5:
            yield f'data: {now}\n\n'
            start_time = now
        else:
            time.sleep(1)  # 1秒待機

eventstream/urls.py

from django.urls import path
from . import views

urlpatterns = [
    path("", views.index, name="index"),
    path('event-stream/', views.event_stream, name='event-stream'),
]

eventstream/templates/eventstream/index.html

<h1>Event Stream Test</h1>

<div>
    <h2 id="id_time"></h2> 
</div>

<script type='text/javascript'>
const evtSource = new EventSource("event-stream");
evtSource.onmessage = function(event){
     //イベントを受信した際の処理
     var timeField = document.getElementById('id_time');
     timeField.innerHTML = event.data;
}
</script>