チュートリアル:ビデオの配信#

このチュートリアルでは、非常に基本的なビデオサーバーを構築します。ビデオを直接配信します。

このチュートリアルは、Quartで条件付きレスポンスを使用して大きなファイルを配信する方法を紹介することを目的としています。完成したコードを見たい場合は、Githubにあります。

1:プロジェクトの作成#

ビデオサーバー用のプロジェクトを作成する必要があります。私はPoetryを使うのが好きです。Poetryはpip(またはBrew)でインストールします。

pip install poetry

Poetryを使って新しいビデオプロジェクトを作成できます。

poetry new --src video

これで、 *video* ディレクトリでプロジェクトを開発できるようになりました。以降のコマンドはすべて *video* ディレクトリで実行する必要があります。

2:依存関係の追加#

このシンプルなビデオサーバーを構築するにはQuartのみが必要です。Quartは、次のコマンドを実行してプロジェクトの依存関係としてインストールできます。

poetry add quart

Poetryは、次のコマンドを実行することで、この依存関係が存在し、パスが正しいことを確認します。

poetry install

3:アプリの作成#

WebサーバーとなるQuartアプリが必要です。これは、 *src/video/__init__.py* に以下を追加することで作成されます。

src/video/__init__.py#
from quart import Quart

app = Quart(__name__)

def run() -> None:
    app.run()

アプリを簡単に実行できるように、 *pyproject.toml* に以下を追加して、poetryスクリプトからrunメソッドを呼び出すことができます。

pyproject.toml#
[tool.poetry.scripts]
start = "video:run"

これにより、次のコマンドでアプリを起動できます。

poetry run start

4:UIの配信#

ユーザーがWebサイトにアクセスした際には、直接配信されるビデオとチャンクで配信されるビデオを同じように表示します。次のHTMLテンプレートを *src/video/templates/index.html* に追加する必要があります。

src/video/templates/index.html#
<video controls width="100%">
  <source src="/video.mp4" type="video/mp4">
</video>

これは、スタイリングの面では非常に基本的なUIです。

これで、ルートパス、つまり / に対してこのテンプレートを配信できます。 *src/video/__init__.py* に以下を追加します。

from quart import render_template

@app.get("/")
async def index():
    return await render_template("index.html")

5:ルートの実装#

大きなファイルを配信するため、条件付きレスポンスを許可する必要があります。これは、レスポンスで返されるデータがリクエストで要求された内容に依存する場合です。これは、 request.range 属性で検査できる Range ヘッダーフィールドを介して行われます。

Quartには、リクエストの範囲に基づいてレスポンスを条件付きにするための組み込みメソッドがあります。1つ目は、ファイルを送信するときにconditional引数を使用する方法、2つ目は、レスポンスの make_conditional メソッドを使用する方法です。前者を以下に示します。これを *src/video/__init__.py* に追加する必要があります。

src/video/__init__.py#
@app.route("/video.mp4")
async def auto_video():
    return await send_file(app.static_folder / "video.mp4", conditional=True)

6:テスト#

アプリをテストするには、条件付き範囲リクエストが行われない限り、ビデオ全体が返されることを確認する必要があります。これは、 *tests/test_video.py* に以下を追加することで行われます。

tests/test_video.py#
from video import app

async def test_auto_video() -> None:
    test_client = app.test_client()
    response = await test_client.get("/video.mp4")
    data = await response.get_data()
    assert len(data) == 255_849

    response = await test_client.get("/video.mp4", headers={"Range": "bytes=200-1000"})
    data = await response.get_data()
    assert len(data) == 801

テストは非同期関数であるため、次のコマンドを実行してpytest-asyncioをインストールする必要があります。

poetry add --dev pytest-asyncio

インストールしたら、 *pyproject.toml* に以下を追加して設定する必要があります。

[tool.pytest.ini_options]
asyncio_mode = "auto"

最後に、次のコマンドでテストを実行できます。

poetry run pytest tests/

Quartのサンプルフォルダでこれを実行している場合は、pytestがQuartのpytest設定を使用しないようにするために、 -c pyproject.toml オプションを追加する必要があります。

7:まとめ#

クライアントのリクエストに応じて、最大部分サイズを制限する機能を含め、大きなファイルを条件付きで配信するサーバーを構築しました。