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

Azure Function プロジェクトを作成

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

1. コマンド ラインで、プロジェクトのディレクトリを作成します。
2. プロジェクト ディレクトリに移動します。
3. 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

2. 次のコードを追加して関数 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 を追加します

@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 プロジェクトを作成します。

1. コマンド ラインで、プロジェクトのディレクトリを作成します。
2. プロジェクト ディレクトリに移動します。
3. 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 によって既定の関数ソース ファイルが作成されて、関数にちなんだ名前のフォルダーに格納されます。 ファイルを必要に応じて編集し、既定のコードをアプリ コードに置き換えます。

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

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

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

func new -n index -t HttpTrigger

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

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

1. 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')

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

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

func new -n negotiate -t HttpTrigger

1. 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"
    }
  ]
}

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

import azure.functions as func


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

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

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

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

1. 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"
    }
  ]
}

1. 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) ]
    }))