Python を使用した Azure Functions: トリガー

Python 開発者は、Azure Functions を使用して、軽量でスケーラブルで効率的なサーバーレス アプリケーションを作成できます。この投稿では、トリガーに焦点を当てます。

Azure Functions のトリガーとは何ですか?

トリガーは Azure Functions の基礎です。これらは、関数の呼び出し方法を決定します。各関数にはトリガーが 1 つだけ必要であり、トリガーのタイプによって関数で使用できるデータ ペイロードが決まります。 Azure は、次のようなさまざまなトリガーをサポートしています。

1. HTTPトリガー

• HTTP リクエスト経由で関数を呼び出すことができます。
• API の構築や Webhook への応答に役立ちます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.route(route="http_trigger", auth_level=func.AuthLevel.ANONYMOUS)
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    return func.HttpResponse("Hello world from HTTP trigger")

パラメータ:

• route: HTTP トリガーが応答する URL パスを指定します。この場合、関数は /api/http_trigger.
でアクセスできます。
• auth_level: 関数の認証レベルを決定します。オプションには次のものが含まれます。
  • 匿名: 認証は必要ありません。
  • 機能: 機能固有のキーが必要です。
  • ADMIN: 管理者レベルのキーが必要です。

2.タイマートリガー

• スケジュールに基づいて機能を実行します。
• Cron 式はスケジュールに使用されます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.timer_trigger(schedule="0 */5 * * * *", arg_name="myTimer", run_on_startup=False, use_monitor=False)
def timer_trigger(myTimer: func.TimerRequest) -> None:

    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function executed.')

パラメータ:

• schedule: CRON 式を使用してスケジュールを定義します。ここで、 0 */5 * * * * は、関数が 0 秒目から開始して 5 分ごとに実行されることを指定します。
• arg_name: TimerRequest オブジェクトを表す、関数に渡される引数の名前。
• run_on_startup: True に設定すると、アプリの起動時に関数がすぐに実行されます。デフォルトは False です。
• use_monitor: Azure がスケジュールの実行漏れを監視するかどうかを決定します。 True の場合、Azure は失敗した実行を確実に再試行します。デフォルトは True です。この例では、False に設定されています。

3.ブロブトリガー

• Azure Blob Storage の変更 (ファイルのアップロードなど) に応答します。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.blob_trigger(arg_name="myblob", path="blobname", connection="BlobStorageConnectionString")
def BlobTrigger(myblob: func.InputStream):
    logging.info(f"Python blob trigger function processed blob"
                f"Name: {myblob.name}"
                f"Blob Size: {myblob.length} bytes")

パラメータ:

• arg_name: BLOB データを表す関数の引数の名前を指定します。これが myblob です。
• path: 関数がリッスンする Blob Storage コンテナー内のパス。この例では、blobname です。
• connection: Blob Storage アカウントの接続文字列を含むアプリケーション設定の名前を参照します。ここでは BlobStorageConnectionString です。

4.キュートリガー

• Azure Storage Queue に追加されたメッセージによってトリガーされます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.route(route="http_trigger", auth_level=func.AuthLevel.ANONYMOUS)
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    return func.HttpResponse("Hello world from HTTP trigger")

パラメータ:

• arg_name: 関数内のキュー メッセージを表す引数の名前を指定します。こちら、アズキューです
• queue_name: 関数がリッスンする Azure ストレージ キューの名前。この場合、それはキュー名です。
• connection: Azure Storage Queue の接続文字列を含むアプリケーション設定を指します。ここでは、QueueConnectionString です。

5.イベントハブトリガー

• Azure Event Hub に送信されたイベントによってトリガーされます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.timer_trigger(schedule="0 */5 * * * *", arg_name="myTimer", run_on_startup=False, use_monitor=False)
def timer_trigger(myTimer: func.TimerRequest) -> None:

    if myTimer.past_due:
        logging.info('The timer is past due!')

    logging.info('Python timer trigger function executed.')

パラメータ:

• arg_name: これは、関数内でイベント データを受け取るパラメーターの名前を指定します。この場合、azeventhub は受信 EventHubEvent を表す変数になります。
• event_hub_name: これは、関数がリッスンしているイベント ハブの名前を示します。 eventhubname をイベント ハブの実際の名前に置き換えます。
• connection: これは、イベント ハブの接続文字列を含むアプリケーション設定の名前を指します。 Azure Function App の設定に、適切な接続文字列値を含む EventHubConnectionString という名前のエントリが含まれていることを確認してください。

6. ServiceBus キュー トリガー

• Azure Service Bus キューに追加されたメッセージによってトリガーされます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.blob_trigger(arg_name="myblob", path="blobname", connection="BlobStorageConnectionString")
def BlobTrigger(myblob: func.InputStream):
    logging.info(f"Python blob trigger function processed blob"
                f"Name: {myblob.name}"
                f"Blob Size: {myblob.length} bytes")

パラメータ:

• arg_name: これは、関数内でメッセージ データを受け取るパラメーターの名前を指定します。この場合、azservicebus は受信 ServiceBusMessage を表す変数になります。
• queue_name: これは、関数がリッスンしている Service Bus キューの名前を示します。 servicebusqueuename を Service Bus キューの実際の名前に置き換えます。
• connection: これは、Service Bus の接続文字列を含むアプリケーション設定の名前を指します。 Azure Function App の設定に、適切な接続文字列値を含む ServiceBusConnectionString という名前のエントリが含まれていることを確認してください。

7. ServiceBus トピックのトリガー

• Azure Service Bus トピックにパブリッシュされたメッセージによってトリガーされます。
• 例:

import azure.functions as func
import datetime
import json
import logging

app = func.FunctionApp()

@app.route(route="http_trigger", auth_level=func.AuthLevel.ANONYMOUS)
def http_trigger(req: func.HttpRequest) -> func.HttpResponse:
    logging.info('Python HTTP trigger function processed a request.')

    return func.HttpResponse("Hello world from HTTP trigger")

パラメータ:

• arg_name: 関数内の Service Bus メッセージを表す引数の名前を指定します。ここでは、azservicebus です。
• subscription_name: トリガーがリッスンする Service Bus サブスクリプションの名前。
• topic_name: トリガーがリッスンする Service Bus トピックの名前。この例では、servicebustopicname.
です。
• connection: Azure Service Bus 名前空間の接続文字列を含むアプリケーション設定を指します。ここでは、ServiceBusConnectionString です。

その他のトリガー

• Cosmos DB トリガー: 変更フィード メカニズムを利用して、Azure Cosmos DB データベースの変更 (挿入と更新) に応答します。
• Dapr Publish Output Binding: 関数が実行中にメッセージを Dapr トピックに発行できるようにし、マイクロサービス間の通信を容易にします。
• Dapr サービス呼び出しトリガー: 他の Dapr 対応サービスによって関数を直接呼び出せるようにし、サービス間通信をサポートします。
• Dapr トピック トリガー: Dapr のパブリッシュ/サブスクライブ メッセージング パターンを介して特定のトピックにパブリッシュされたメッセージに応答して関数を実行します。
• Event Grid トリガー: イベントが Azure Event Grid トピックに送信されるときに関数をアクティブ化し、リアクティブなイベント駆動型アーキテクチャを可能にします。

以上がPython を使用した Azure Functions: トリガーの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。