Azure Service Bus Topic を Python で送受信する

この記事の目的

Azure Service Bus Topic をローカルPython環境から送信し、Azure Functions で受信するまでの手順を解説します。

👇これより先は下記内容を前提とします

• Azure Functions でPython関数を作成してデプロイする

Service Bus Topic 関連リソースを作成する

Azure ログインを実施し、サブスクリプションを設定します。

az login
az account set --subscription "*****-****-****-****-****"

リソースグループresource_group_for_pythonに、Service Bus メッセージング名前空間servicebus111を作成します。

az servicebus namespace create^
 --resource-group resource_group_for_python^
 --name servicebus111^
 --location japaneast

Service Bus Topic MyTopic111を作成します。

az servicebus topic create^
 --resource-group resource_group_for_python^
 --namespace-name servicebus111^
 --name MyTopic111

サブスクリプション subscription01を作成します。

az servicebus topic subscription create^
 --resource-group resource_group_for_python^
 --namespace-name servicebus111^
 --topic-name MyTopic111^
 --name subscription01

関数用の関連 Azure リソースを作成する

Azure ログインを実施し、サブスクリプションを設定します。

az login
az account set --subscription "*****-****-****-****-****"

リソースグループresource_group_for_pythonに、ストレージアカウントsasample111を作成します。

az storage account create^
 --name sasample111^
 --location japaneast^
 --resource-group resource_group_for_python^
 --sku Standard_LRS

リソースグループresource_group_for_pythonに、関数アプリappsample111を作成します。
Pythonバージョンは本記事記載時点の最新バージョン3.11を指定します。

az functionapp create^
 --resource-group resource_group_for_python^
 --runtime python^
 --runtime-version 3.11^
 --functions-version 4^
 --os-type linux^
 --consumption-plan-location japaneast^
 --storage-account sasample111^
 --name appsample111

関数アプリappsample111にシステム割り当て ID を追加します。

az webapp identity assign^
 --name appsample111^
 --resource-group resource_group_for_python

下記のような出力が得られます。
(principalIdが関数アプリappsample111のシステム割り当てID)

{
 "principalId": "00000000-0000-0000-0000-000000000000",
 "tenantId": "11111111-1111-1111-1111-111111111111",
 "type": "SystemAssigned",
 "userAssignedIdentities": null
}

下記コマンドで関数アプリappsample111にアプリケーション設定ServiceBusConnection__fullyQualifiedNamespaceを追加し、Service Bus 名前空間servicebus112.servicebus.windows.netを設定します。

az functionapp config appsettings set^
 --name appsample111^
 --resource-group resource_group_for_python^
 --settings ServiceBusConnection__fullyQualifiedNamespace="servicebus111.servicebus.windows.net"

ロールを割り当てる

ロール一覧を取得します。

az role definition list --output table

下記のような出力が得られます。

Name                                                        Type                                     Description
----------------------------------------------------------  ---------------------------------------  -----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
AcrPush                                                     Microsoft.Authorization/roleDefinitions  acr push
API Management Service Contributor                          Microsoft.Authorization/roleDefinitions  Can manage service and the APIs
AcrPull                                                     Microsoft.Authorization/roleDefinitions  acr pull
・・・・
Azure Service Bus Data Owner                                Microsoft.Authorization/roleDefinitions  Allows for full access to Azure Service Bus resources.
Azure Event Hubs Data Owner                                 Microsoft.Authorization/roleDefinitions  Allows for full access to Azure Event Hubs resources.

Service Bus Data Owner権限を割り当てます。

az role assignment create^
 --assignee "00000000-0000-0000-0000-000000000000"^
 --role "Azure Service Bus Data Owner"^
 --scope "/subscriptions/***-***-***-***/resourceGroups/resource_group_for_python/providers/Microsoft.ServiceBus/namespaces/servicebus112"

ローカル関数の作成

func initコマンドを実行して、LocalFunctionSurviceBusTopicProj というフォルダーに関数プロジェクトを作成します。

func init LocalFunctionSurviceBusTopicProj --python -m V2

プロジェクトフォルダーに移動します。

cd LocalFunctionSurviceBusTopicProj

関数を作成します。

func new

下記のようにtemplateの選択プロンプトが表示されるので、ここではServiceBus Topic triggerを選択します。

Use the up/down arrow keys to select a template:
Blob trigger
CosmosDB trigger
EventGrid trigger
EventHub trigger
HTTP trigger
Queue trigger
ServiceBus Queue trigger
ServiceBus Topic trigger
Timer Trigger

下記内容を指定します。

• Function Name:servicebus_topic_func01
• Service Bus Topic Name:上記で作成したトピック名MyTopic111
• Service Bus Subscripton Name:上記で作成したサブスクリプション名subscription01
• Service Bus Connection:ServiceBusConnection

最終的に下記内容が出力されます。

Use the up/down arrow keys to select a template:Function Name: [servicebus_topic_trigger] servicebus_topic_func01
Service Bus Topic Name: [servicebustopicname] MyTopic111
Service Bus Subscripton Name: [servicebussubscriptionName] subscription01
Service Bus Connection: [ServiceBusConnectionString] ServiceBusConnection
Appending to c:\tmp\LocalFunctionSurviceBusTopicProj\function_app.py
The function "servicebus_topic_func01" was created successfully from the "ServiceBus Topic trigger" template.

Azure の関数アプリappsample111に関数プロジェクトをデプロイします。

func azure functionapp publish appsample111

メッセージをAzure Service Busトピックに送信する

メッセージをAzure Service Busトピックに送信するサンプルコードです。

import json

from azure.servicebus import ServiceBusClient
from azure.servicebus import ServiceBusMessage
from azure.identity import DefaultAzureCredential

# Service Bus 名前空間
FULLY_QUALIFIED_NAMESPACE = "servicebus111.servicebus.windows.net"

# トピック名
TOPIC_NAME = "MyTopic111"

credential = DefaultAzureCredential()

def send_message():
    print("run() start")

    servicebus_client = ServiceBusClient(
        fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
        credential=credential,
        logging_enable=True)
    
    sender = servicebus_client.get_topic_sender(topic_name=TOPIC_NAME)

    send_param = {'param1':'value1'}
    
    message = ServiceBusMessage(body=json.dumps(send_param))

    # send the message to the topic
    sender.send_messages(message)

print("start")
send_message()
print("end")

👇 上記コードの実行後に、関数アプリがトピックを受信していることを確認できます。

👇関連記事

• Azure Functions でPython関数を作成してデプロイする
• ローカル開発環境で Azure SDK for Python を利用する

👇参考URL

• Azure CLI を使用して Service Bus トピックとそのサブスクリプションを作成する
• アプリケーション設定の反映
• Azure Service Bus トピックへのメッセージ送信とトピックのサブスクリプションからのメッセージ受信 (Python)
• Azure Functions の Azure Service Bus トリガー
• Azure Functions における Azure Service Bus のバインド
• App Service と Azure Functions でマネージド ID を使用する方法

[keywords]
Azure Service Bus Topic Python

Azure Service Bus Topic を Python で送受信する

更新日:2024/02/02