クイックスタート: Azure Function と Azure SignalR Service を Python で使用してサーバーレスアプリを作成する

[アーティクル]
12/16/2024

Azure Functions と Python を使用して、クライアントにメッセージをブロードキャストするサーバーレスアプリケーションを構築することで、Azure SignalR Service の使用を開始します。ローカル環境で関数を実行し、クラウド内の Azure SignalR Service インスタンスに接続します。このクイックスタートを完了すると、ご利用の Azure アカウントでわずかな (数セント未満の) コストが発生します。

注意

この記事のるコードは、GitHub から入手できます。

重要

この記事に示す生の接続文字列は、デモンストレーションのみを目的としています。

接続文字列には、アプリケーションが Azure Web PubSub サービスにアクセスするために必要な認可情報が含まれています。接続文字列内のアクセスキーは、サービスのルートパスワードに似ています。運用環境では、常にアクセスキーを保護してください。 Azure Key Vault を使用してキーの管理とローテーションを安全に行い、Microsoft Entra ID を使用して接続文字列をセキュリティで保護し、Microsoft Entra ID を使用してアクセスを認可します。

アクセスキーを他のユーザーに配布したり、ハードコーディングしたり、他のユーザーがアクセスできるプレーンテキストで保存したりしないでください。キーが侵害された可能性があると思われる場合は、キーをローテーションしてください。

前提条件

このクイックスタートは、macOS、Windows、または Linux で実行できます。以下のものが必要になります。

前提条件	説明
Azure サブスクリプション	Azure サブスクリプションをお持ちでない場合は、Azure の無料アカウントを作成してください
コードエディター	Visual Studio Code などのコードエディターが必要になります。
Azure Functions Core Tools	Python Azure Function アプリをローカルで実行するには、バージョン 2.7.1505 以降が必要です。
Python 3.7 以降	Azure Functions には Python 3.7 以降が必要です。サポートされている Python のバージョンを参照してください。
Azurite	SignalR バインディングに Azure Storage が必要です。関数がローカルで動作している場合は、ローカルストレージエミュレーターを使用できます。
Azure CLI	必要に応じて、Azure CLI を使用して Azure SignalR Service インスタンスを作成できます。

Azure SignalR Service のインスタンスを作成する

このセクションでは、アプリに使う基本的な Azure SignalR インスタンスを作成します。次の手順では、Azure portal を使って新しいインスタンスを作成しますが、Azure CLI を使うこともできます。詳細については、Azure SignalR Service CLI リファレンスの az signalr create コマンドを参照してください。

Azure portal にサインインします。
ページの左上にある [+ リソースの作成] を選択します。
[リソースの作成] ページで、[Search services and marketplace](サービスとマーケットプレースを検索) テキストボックスに「signalr」と入力し、リストから [SignalR Service] を選びます。
[SignalR Service] ページで [作成] を選びます。
[基本] タブで、新しい SignalR Service のインスタンスに必要な情報を入力します。次の値を入力します。

フィールド	推奨値	説明
サブスクリプション	サブスクリプションの選択	新しい SignalR Service インスタンスを作成するために使うサブスクリプションを選びます。
リソースグループ	SignalRTestResources というリソースグループを作成する	SignalR リソースのリソースグループを選択または作成します。既存のリソースグループを使う代わりに、このチュートリアルのために新しいリソースグループを作成すると便利です。チュートリアルの終了後にリソースを解放するには、リソースグループを削除します。リソースグループを削除すると、そのグループに所属するすべてのリソースも削除されます。この削除操作は元に戻すことができません。保存するリソースが含まれていないことを確認してから、リソースグループを削除してください。詳細については、リソースグループを使用した Azure リソースの管理に関するページを参照してください。
リソース名	testsignalr	SignalR リソースに使用する一意のリソース名を入力します。 testsignalr がお使いのリージョンに既に導入されている場合は、数字や文字を追加して名前が一意になるようにします。名前は 1 ～ 63 文字の文字列で、数字、英字、ハイフン (`-`) 文字のみを使用する必要があります。名前の先頭と末尾にはハイフン文字を使用できません。また、連続するハイフン文字は無効です。
[リージョン]	自分のリージョンを選択します	新しい SignalR Service インスタンスに適切なリージョンを選びます。 Azure SignalR Service は、現在すべてのリージョンで使用できるわけではありません。詳細については、Azure SignalR Service リージョンの可用性に関するページを参照してください
価格レベル	[変更] を選び、[Free (Dev/Test Only)](無料 (Dev/Test のみ)) を選びます。 [選択] を選び、価格レベルを確認します。	Azure SignalR Service には、Free、Standard、Premium という 3 つの価格レベルがあります。チュートリアルでは、前提条件で特に明記されない限り、[Free] レベルを使います。レベルと価格による機能の違いの詳細については、「Azure SignalR Service の価格」を参照してください
サービスモード	適切なサービスモードを選びます	Web アプリで SignalR のハブロジックをホストしていて、SignalR Service をプロキシとして使う場合は、[既定] を使います。 Azure Functions などのサーバーレステクノロジを使って SignalR のハブロジックをホストする場合は、[サーバーレス] を使います。 [クラシック] モードは下位互換性のためだけにあり、使わないことをお勧めします。詳細については、「Azure SignalR Service のサービスモード」を参照してください。

SignalR チュートリアルでは、[ネットワーク] と [タグ] タブの設定を変更する必要はありません。

[基本] タブの下部にある [確認と作成] ボタンを選びます。
[確認と作成] タブで、値を確認し [作成] を選びます。デプロイが完了するまでしばらくかかります。
デプロイが完了したら、[リソースに移動] ボタンを選びます。
SignalR リソースページで、左側のメニューから、[設定] の下にある [キー] を選びます。
主キーの [接続文字列] をコピーします。この接続文字列は、このチュートリアルで後ほどアプリを構成するために必要です。

Azure Function プロジェクトを作成

ローカルの Azure Function プロジェクトを作成します。

コマンドラインで、プロジェクトのディレクトリを作成します。
プロジェクトディレクトリに移動します。
Azure Function の func init コマンドを使用して関数プロジェクトを初期化します。

# Initialize a function project
func init --worker-runtime python

関数を作成する

プロジェクトを初期化した後、関数を作成する必要があります。このプロジェクトには、次の 3 つの関数が必要です。

index: クライアントの Web ページをホストします。
negotiate: クライアントがアクセストークンを取得できるようにします。
broadcast: タイムトリガーを使用して、すべてのクライアントにメッセージを定期的にブロードキャストします。

プロジェクトのルートディレクトリから func new コマンドを実行すると、Azure Functions Core Tools によって function_app.py ファイルに関数コードが追加されます。必要に応じて、既定のコードをアプリコードに置き換えて、パラメーターと内容を編集します。

インデックス関数を作成する

このサンプル関数を、自分の関数のテンプレートとして使用できます。

function_app.py ファイルを開きます。このファイルに関数が含まれます。まず、必要な import ステートメントを含むようにファイルを変更し、次の関数で使用するグローバル変数を定義します。

import azure.functions as func
import os
import requests
import json 

app = func.FunctionApp()

etag = ''
start_count = 0

次のコードを追加して関数 index を追加します

@app.route(route="index", auth_level=func.AuthLevel.ANONYMOUS)
def index(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

この関数は、クライアントの Web ページをホストします。

ネゴシエート関数を作成する

次のコードを追加して関数 negotiate を追加します。

この記事に示す生の接続文字列は、デモンストレーションのみを目的としています。運用環境では、常にアクセスキーを保護してください。 Azure Key Vault を使用してキーの管理とローテーションを安全に行い、Microsoft Entra ID を使用して接続文字列をセキュリティで保護し、Microsoft Entra ID を使用してアクセスを認可します。

@app.route(route="negotiate", auth_level=func.AuthLevel.ANONYMOUS, methods=["POST"])
@app.generic_input_binding(arg_name="connectionInfo", type="signalRConnectionInfo", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def negotiate(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

この関数により、クライアントはアクセストークンを取得できるようになります。

ブロードキャスト関数を作成します。

次のコードを追加して関数 broadcast を追加します

@app.timer_trigger(schedule="*/1 * * * *", arg_name="myTimer",
              run_on_startup=False,
              use_monitor=False)
@app.generic_output_binding(arg_name="signalRMessages", type="signalR", hubName="serverless", connectionStringSetting="AzureSignalRConnectionString")
def broadcast(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-functions-python-worker', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://api.github.com/repos/azure/azure-functions-python-worker is: ' + str(start_count) ]
    }))

この関数はタイムトリガーを使用して、すべてのクライアントにメッセージを定期的にブロードキャストします。

Azure Function プロジェクトを作成

ローカルの Azure Function プロジェクトを作成します。

コマンドラインで、プロジェクトのディレクトリを作成します。
プロジェクトディレクトリに移動します。
Azure Function の func init コマンドを使用して関数プロジェクトを初期化します。

# Initialize a function project
func init --worker-runtime python --model v1

関数を作成する

プロジェクトを初期化した後、関数を作成する必要があります。このプロジェクトには、次の 3 つの関数が必要です。

index: クライアントの Web ページをホストします。
negotiate: クライアントがアクセストークンを取得できるようにします。
broadcast: タイムトリガーを使用して、すべてのクライアントにメッセージを定期的にブロードキャストします。

プロジェクトのルートディレクトリから func new コマンドを実行すると、Azure Function の Core Tools によって既定の関数ソースファイルが作成されて、関数にちなんだ名前のフォルダーに格納されます。ファイルを必要に応じて編集し、既定のコードをアプリコードに置き換えます。

インデックス関数を作成する

このサンプル関数を、自分の関数のテンプレートとして使用できます。

次のコマンドを実行して index 関数を作成します。

func new -n index -t HttpTrigger

index/function.json を編集して内容を次の json コードに置き換えます。

{
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "get",
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    }
  ]
}

index/_init_.py を編集して内容を次のコードに置き換えます。

import os

import azure.functions as func

def main(req: func.HttpRequest) -> func.HttpResponse:
    f = open(os.path.dirname(os.path.realpath(__file__)) + '/../content/index.html')
    return func.HttpResponse(f.read(), mimetype='text/html')

ネゴシエート関数を作成する

次のコマンドを実行して negotiate 関数を作成します。

func new -n negotiate -t HttpTrigger

negotiate/function.json を編集して内容を次の json コードに置き換えます。

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "authLevel": "anonymous",
      "type": "httpTrigger",
      "direction": "in",
      "name": "req",
      "methods": [
        "post"
      ]
    },
    {
      "type": "http",
      "direction": "out",
      "name": "$return"
    },
    {
      "type": "signalRConnectionInfo",
      "name": "connectionInfo",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "in"
    }
  ]
}

negotiate/_init_.py を編集して内容を次のコードに置き換えます。

import azure.functions as func


def main(req: func.HttpRequest, connectionInfo) -> func.HttpResponse:
    return func.HttpResponse(connectionInfo)

ブロードキャスト関数を作成します。

次のコマンドを実行して broadcast 関数を作成します。

func new -n broadcast -t TimerTrigger
# install requests
pip install requests

broadcast/function.json を編集して内容を次のコードに置き換えます。

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "myTimer",
      "type": "timerTrigger",
      "direction": "in",
      "schedule": "*/5 * * * * *"
    },
    {
      "type": "signalR",
      "name": "signalRMessages",
      "hubName": "serverless",
      "connectionStringSetting": "AzureSignalRConnectionString",
      "direction": "out"
    }
  ]
}

broadcast/_init_.py を編集して内容を次のコードに置き換えます。

import requests
import json

import azure.functions as func

etag = ''
start_count = 0

def main(myTimer: func.TimerRequest, signalRMessages: func.Out[str]) -> None:
    global etag
    global start_count
    headers = {'User-Agent': 'serverless', 'If-None-Match': etag}
    res = requests.get('https://api.github.com/repos/azure/azure-signalr', headers=headers)
    if res.headers.get('ETag'):
        etag = res.headers.get('ETag')

    if res.status_code == 200:
        jres = res.json()
        start_count = jres['stargazers_count']

    signalRMessages.set(json.dumps({
        'target': 'newMessage',
        'arguments': [ 'Current star count of https://github.com/Azure/azure-signalr is: ' + str(start_count) ]
    }))

index.html ファイルを作成する

このアプリのクライアントインターフェイスは Web ページです。 index 関数は content/index.html ファイルから HTML コンテンツを読み取ります。

content という名前のフォルダーをプロジェクトルートフォルダーに作成します。
content/index.html ファイルを作成します。
次のコンテンツを content/index.html ファイルにコピーして保存します。

<html>

<body>
  <h1>Azure SignalR Serverless Sample</h1>
  <div id="messages"></div>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/microsoft-signalr/3.1.7/signalr.min.js"></script>
  <script>
    let messages = document.querySelector('#messages');
    const apiBaseUrl = window.location.origin;
    const connection = new signalR.HubConnectionBuilder()
        .withUrl(apiBaseUrl + '/api')
        .configureLogging(signalR.LogLevel.Information)
        .build();
      connection.on('newMessage', (message) => {
        document.getElementById("messages").innerHTML = message;
      });

      connection.start()
        .catch(console.error);
  </script>
</body>

</html>

SignalR Service 接続文字列を関数アプリの設定に追加する

最後の手順では、 Azure Function アプリ設定に SignalR Service 接続文字列を設定します。

Azure portal で、先ほどデプロイした SignalR インスタンスに移動します。
[Key](キー) を選択し、SignalR Service インスタンスの接続文字列を表示します。
プライマリ接続文字列をコピーし、コマンドを実行します。
```
func settings add AzureSignalRConnectionString "<signalr-connection-string>"
```

Azure Function アプリをローカルで実行します

Azurite ストレージエミュレーターを起動します。

azurite

Azure Function アプリをローカル環境で実行します。

func start

注意

BLOB ストレージで読み取りエラーを示すエラーが表示された場合は、local.settings.json ファイルの 'AzureWebJobsStorage' 設定が UseDevelopmentStorage=true に設定されていることを確認します。

Azure Function がローカルで実行された後、http://localhost:7071/api/index に移動します。ページには、GitHub Azure/azure-signalr リポジトリの現在の星数が表示されます。 GitHub でリポジトリに星を付けたり外したりすると、数秒ごとに更新された数が表示されます。

リソースをクリーンアップする

このアプリの使用を続けない場合は、次の手順に従って、このクイックスタートで作成したすべてのリソースを削除して、課金が発生しないようにします。

Azure Portal の左端で [リソースグループ] を選択し、作成したリソースグループを選択します。検索ボックスを使用して名前でリソースグループを検索することもできます。
表示されたウィンドウでリソースグループを選択し、[リソースグループの削除] をクリックします。
新しいウィンドウで、削除するリソースグループの名前を入力し、[削除] をクリックします。

問題がある場合は、トラブルシューティングガイドをお試しになるか、ご連絡ください。

次のステップ

このクイックスタートでは、リアルタイムのサーバーレスアプリケーションを構築してローカルで実行しました。次に、SignalR Service を使用して、クライアントと Azure Functions の間で双方向通信を使用する方法について詳しく説明します。

Azure Functions における SignalR サービスのバインド

サーバーレスでの双方向通信

VS Code で Azure Functions をデプロイする

次の方法で共有

クイックスタート: Azure Function と Azure SignalR Service を Python で使用してサーバーレスアプリを作成する

前提条件

Azure SignalR Service のインスタンスを作成する

Azure Function プロジェクトを作成

関数を作成する

インデックス関数を作成する

ネゴシエート関数を作成する

ブロードキャスト関数を作成します。

Azure Function プロジェクトを作成

関数を作成する

インデックス関数を作成する

ネゴシエート関数を作成する

ブロードキャスト関数を作成します。

index.html ファイルを作成する

SignalR Service 接続文字列を関数アプリの設定に追加する

Azure Function アプリをローカルで実行します

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

次の方法で共有

クイックスタート: Azure Function と Azure SignalR Service を Python で使用してサーバーレス アプリを作成する

前提条件

Azure SignalR Service のインスタンスを作成する

Azure Function プロジェクトを作成

関数を作成する

インデックス関数を作成する

ネゴシエート関数を作成する

ブロードキャスト関数を作成します。

Azure Function プロジェクトを作成

関数を作成する

インデックス関数を作成する

ネゴシエート関数を作成する

ブロードキャスト関数を作成します。

index.html ファイルを作成する

SignalR Service 接続文字列を関数アプリの設定に追加する

Azure Function アプリをローカルで実行します

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

クイックスタート: Azure Function と Azure SignalR Service を Python で使用してサーバーレスアプリを作成する