Python을 사용한 Azure Functions: 트리거

Python 개발자는 Azure Functions를 사용하여 가볍고 확장 가능하며 효율적인 서버리스 애플리케이션을 만들 수 있습니다. 이번 게시물에서는 트리거에 중점을 두겠습니다.

Azure Functions의 트리거란 무엇입니까?

트리거는 Azure Functions의 기초입니다. 함수가 호출되는 방법을 결정합니다. 각 함수에는 정확히 하나의 트리거가 있어야 하며, 트리거 유형에 따라 함수에 사용할 수 있는 데이터 페이로드가 결정됩니다. Azure는 다음을 포함한 다양한 트리거를 지원합니다.

1. HTTP 트리거

• HTTP 요청을 통해 함수를 호출할 수 있습니다.
• API를 구축하거나 웹훅에 응답하는 데 유용합니다.
• 예:

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

매개변수:

• 일정: 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입니다.
• 경로: 함수가 수신 대기하는 Blob Storage 컨테이너의 경로입니다. 이 예에서는 blobname입니다.
• 연결: Blob Storage 계정에 대한 연결 문자열이 포함된 애플리케이션 설정의 이름을 나타냅니다. 여기는 BlobStorageConnectionString입니다.

4. 대기열 트리거

• Azure Storage 대기열에 추가된 메시지에 의해 트리거됩니다.
• 예:

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 Storage 큐의 이름입니다. 이 경우에는 queuename입니다.
• 연결: Azure Storage 대기열에 대한 연결 문자열이 포함된 애플리케이션 설정을 나타냅니다. 여기는 QueueConnectionString입니다.

5. EventHub 트리거

• 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을 Event Hub의 실제 이름으로 바꿉니다.
• 연결: 이벤트 허브에 대한 연결 문자열이 포함된 애플리케이션 설정의 이름을 나타냅니다. Azure 함수 앱의 설정에 적절한 연결 문자열 값과 함께 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 함수 앱의 설정에 적절한 연결 문자열 값과 함께 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입니다.
• 연결: Azure Service Bus 네임스페이스에 대한 연결 문자열이 포함된 애플리케이션 설정을 나타냅니다. 여기는 ServiceBusConnectionString입니다.

기타 트리거

• Cosmos DB 트리거: 변경 피드 메커니즘을 활용하여 Azure Cosmos DB 데이터베이스의 변경 사항(삽입 및 업데이트)에 응답합니다.
• Dapr 게시 출력 바인딩: 함수가 실행 중에 Dapr 주제에 메시지를 게시할 수 있도록 하여 마이크로서비스 간 통신을 촉진합니다.
• Dapr 서비스 호출 트리거: 다른 Dapr 지원 서비스에서 직접 기능을 호출하여 서비스 간 통신을 지원합니다.
• Dapr 주제 트리거: Dapr의 게시-구독 메시징 패턴을 통해 특정 주제에 게시된 메시지에 대한 응답으로 기능을 실행합니다.
• Event Grid 트리거: 이벤트가 Azure Event Grid 주제로 전송될 때 기능을 활성화하여 반응형 이벤트 기반 아키텍처를 허용합니다.

위 내용은 Python을 사용한 Azure Functions: 트리거의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!