Python を使用した Azure Functions: トリガー-Python チュートリアル-php.cn

Azure Functions with Python: Triggers

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 サイトの他の関連記事を参照してください。