Django 4.2 で ASGI サポートされた StreamingHTTPResponse を触ってみる

はじめに
環境
サンプルコード
動作確認
- Chrome
- cURL
- Swift
まとめ

はじめに

以前から実装されていた StreamingHTTPResponse ですが、 Django 4.2 からは ASGI で提供された Django において async イテレーターによる実装が可能になりました。*1
以前より、大きな CSV などの大きなデータをレスポンスにする際に使用されるなど *2 で使われていたようです。
私自身、数ヶ月前まで iOS アプリエンジニアをずっとやっていて（今も傍らでやってるんだけれども） Django 2 系あたりで新機能について学習がストップしていました。

近年、生成 AI がブームとなり ChatGPT を始め Claude やなにならたくさんありますね。その中ですごい大量のデータを扱うことが増えてきていると思います。 *3
例に挙げた ChatGPT を使用する際に AI が返してくるときに少しずつ文字が表示されることと思います。
そういう実装をする場合に Django の StreamingHTTPResponse が使用できるとのことなので実装をやってみようという記事です。

環境

Python 3.12.4
Django==4.2.14
uvicorn==0.30.1

サンプルコード

Django の開発環境は本題ではないのですっ飛ばします。
views.py に以下のコードを書きました。

import asyncio

from django.http import StreamingHttpResponse


async def some_streaming_content():
    for index in range(10):
        await asyncio.sleep(1)
        yield index


async def streaming():
    async for chunk in some_streaming_content():
        yield chunk


async def sample(request):
    return StreamingHttpResponse(streaming())

今回、他のパスは実装する必要がないので http://127.0.0.1:8000/ にアクセスしたら sample 関数の実装が呼ばれるように実装しています。ほかの Django プロジェクトが起動しており、落としたりするのが面倒なので今回は 8001 ポートを使用しています。

urlpatterns = [
    path('', sample),
]

実装した Django プロジェクトは python manage.py runserver で起動せずに uvicorn を使用して起動をします。

$ uvicorn sample_proj.asgi:application --reload --port 8000
INFO:     Will watch for changes in these directories: ['/path/sample_proj']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx] using StatReload
INFO:     Started server process [37826]
INFO:     Waiting for application startup.
INFO:     ASGI 'lifespan' protocol appears unsupported.
INFO:     Application startup complete.

動作確認

Chrome

012 まで出てしまっていますがこれは多分仕方ないやつ。
「初期読み込みが完了したら stream を始める」というような実装にしたら順番に表示されるかと思います。

cURL

curl の -N のオプションで Stream ができるということを初めて知りました。

Swift

final class Streaming: NSObject, URLSessionDataDelegate {
  private let yieldHandler: @Sendable (String) -> Void
  private var task: URLSessionTask!

  init(urlRequest: URLRequest, yieldHandler: @Sendable @escaping (String) -> Void) {
    self.yieldHandler = yieldHandler
    super.init()
    let session = URLSession(configuration: .default, delegate: self, delegateQueue: nil)
    self.task = session.dataTask(with: urlRequest)
  }

  deinit {
    task.cancel()
  }

  func start() {
    task.resume()
  }

  func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
    guard let string = String(data: data, encoding: .utf8) else { return }
    yieldHandler(string)
  }

  func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: (any Error)?) {
    self.task.cancel()
  }
}

func handleStream() async throws {
  let stream = AsyncStream<String> { continuation in
    let urlRequest = URLRequest(url: URL(string: "http://127.0.0.1:8001")!)
    let streaming = Streaming(
      urlRequest: urlRequest,
      yieldHandler: { string in
        continuation.yield(string)
        if string == "9" {
          continuation.finish()
        }
      }
    )
    streaming.start()
  }

  var texts: [String] = []
  for await item in stream {
    texts.append(item)
    print(texts.joined(separator: " "))
  }
}

Task {
  try await handleStream()
}