Azure Functions 與 Python：觸發器

Python 開發人員可以使用 Azure Functions 建立輕量級、可擴充且高效的無伺服器應用程式。在這篇文章中，我們將重點放在觸發器上。

Azure Functions 中的觸發器是什麼？

觸發器是 Azure Functions 的基礎。它們決定如何呼叫函數。每個函數必須有一個觸發器，而觸發器類型決定了該函數可用的資料負載。 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 儲存體中的變更（例如檔案上傳）。
• 範例：

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 資料的參數名稱。這是我的blob。
• path: 函數偵聽的 Blob 儲存容器中的路徑。在此範例中，它是 blobname。
• connection: 指包含 Blob 儲存帳戶的連接字串的應用程式設定的名稱。這是 BlobStorageConnectionString。

4.隊列觸發

• 由新增至 Azure 儲存體佇列的訊息觸發。
• 範例：

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：指定函數中代表佇列訊息的參數名稱。在這裡，是 azqueue。
• queue_name：函數偵聽的 Azure 儲存佇列的名稱。在本例中，它是佇列名稱。
• 連線：指包含 Azure 儲存佇列連接字串的應用程式設定。在這裡，它是 QueueConnectionString。

5.事件中心觸發器

• 由傳送到 Azure 事件中心的事件觸發。
• 範例：

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 替換為事件中心的實際名稱。
• 連線： 這是指包含事件中心連接字串的應用程式設定的名稱。確保 Azure Function App 的設定包含名為 EventHubConnectionString 的項目以及對應的連線字串值。

6.服務總線隊列觸發器

• 由新增至 Azure 服務總線佇列的訊息觸發。
• 範例：

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: 這表示函數正在偵聽的服務總線佇列的名稱。將 servicebusqueuename 替換為您的服務總線佇列的實際名稱。
• connection: 這是指包含服務匯流排連接字串的應用程式設定的名稱。確保 Azure Function App 的設定包含名為 ServiceBusConnectionString 的條目以及對應的連接字串值。

7. ServiceBus 主題觸發器

• 由發佈到 Azure 服務總線主題的訊息觸發。
• 範例：

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：指定函數中代表服務總線訊息的參數名稱。在這裡，它是 azservicebus。
• subscription_name：觸發器偵聽的服務總線訂閱的名稱。
• topic_name：觸發器偵聽的服務總線主題的名稱。在此範例中，它是 servicebustopicname。
• 連線：指包含 Azure 服務匯流排命名空間的連線字串的應用程式設定。這裡是ServiceBusConnectionString。

其他觸發因素

• Cosmos DB 觸發器： 利用變更來源機制來回應 Azure Cosmos DB 資料庫中的變更（插入和更新）。
• Dapr 發布輸出綁定： 允許函數在執行期間將訊息發佈到 Dapr 主題，從而促進微服務之間的通訊。
• Dapr 服務呼叫觸發器： 允許其他支援 Dapr 的服務直接呼叫函數，支援服務到服務通訊。
• Dapr 主題觸發器： 執行函數以回應透過 Dapr 的發布-訂閱訊息傳遞模式發佈到特定主題的訊息。
• 事件網格觸發器：當事件傳送到 Azure 事件網格主題時啟動函數，從而實現反應式事件驅動架構。

以上是Azure Functions 與 Python：觸發器的詳細內容。更多資訊請關注PHP中文網其他相關文章！