Azure Functions Python Programming v2 で Blob Storage Trigger (w/ Event Grid) を使う！

本記事について

Azure Functions で Python のコードを動かすには、現在 Programming Model が v1 と v2 の2種類があります。

この記事では、v2 モデルのほうで、Blob Storage Trigger を利用する際に、より高速で高信頼な Event Grid を経由させてトリガーさせるときの設定を見ていきます。

実際の手順

下記に主な手順を見ていきます。

VS Code から Python v2 で Blob Trigger の関数アプリを作成する

VS Code のコマンドパレットで、Azure Functions: Create Function... を選択し、Python v2 で Blob Trigger で関数アプリのコードを生成します。

function_app.py を以下のように書き換えます。

function_app.py

import azure.functions as func
import logging

app = func.FunctionApp()

@app.blob_trigger(arg_name="myblob", path="mycontainer/{name}",
                               connection="Connection_String",
                               source="EventGrid") 
def BlobTriggerEventGrid(myblob: func.InputStream):
    logging.info(f"Python blob trigger function processed blob"
                f"Name: {myblob.name}"
                f"Blob Size: {myblob.length} bytes")

    file = myblob.read()
    logging.info(f"File Content: {file}")

• path はトリガーされるファイルの Blob Storage での PATH になります。上記例は、コンテナー直下にファイルを置く構成です。

• connection は環境変数への参照になります。Azure Functions の環境変数に同名の変数を作成し、今回利用するストレージアカウントの接続文字列を入れておきます。

• source には、source="EventGrid" という値を入れます。

そして、Azure 上の関数アプリにこの関数をデプロイします。

ストレージアカウントでイベントを設定する

次にストレージアカウント側で、イベントを設定します。

イベントサブスクリプション名とシステムトピック名は適当な値を入れます。

フィルターはトリガーで使いたい種別を選択します。

エンドポイントは種類は webhook にし、エンドポイントは下記の値を入力します。

https://<関数アプリ名>.azurewebsites.net/runtime/webhooks/blobs?functionName=Host.Functions.<関数名>&code=<BLOB_EXTENSION_KEY>

関数アプリ名、関数名はそれぞれ作成したものの値を入れ、BLOB_EXTENSION_KEY は関数アプリのアプリキーのページから取得します。

次のフィルタータブでは、イベントをトリガーする対象のファイルを制限できます。例えば、特定のコンテナー内のテキストファイルに絞りたいときは、下記のように記述します。

/blobServices/default/containers/<コンテナー名>

作成すると、Event Grid リソースが作成され、下記のようなイベントサブスクリプションができています。

関数のテスト

この状態で、ストレージアカウント内の mycontainer にテキストファイルをアップロードしてみます。

すると、関数が実行され、下記のようなログが出てきます。

最後に

＊本稿は、個人の見解に基づいた内容であり、所属する会社の公式見解ではありません。また、いかなる保証を与えるものでもありません。正式な情報は、各製品の販売元にご確認ください。