クイック スタート: Azure Service Bus キューとの間でメッセージを送受信する (Python)

自分の Azure アカウントでこのクイックスタートを使用するには:

• 開発者マシンにパスワードレス認証を提供する Azure CLI をインストールします。
• ターミナルまたはコマンド プロンプトで、az login を使って Azure アカウントでサインインします。
• リソースに適切なデータ ロールを追加するときは、同じアカウントを使います。
• 同じターミナルまたはコマンド プロンプトでコードを実行します。
• Service Bus 名前空間のキューの名前を書き留めてください。 これはコード内で必要です。

コードで使用する次の値に注意してください。

• Service Bus 名前空間の接続文字列
• 作成した Service Bus 名前空間のキュー

Microsoft Entra ユーザーにロールを割り当てる

ローカルで開発するときは、Azure Service Bus に接続するユーザー アカウントに適切なアクセス許可があることを確認します。 メッセージを送受信するには、 Azure Service Bus データ所有者 ロールが必要です。 このロールを自分に割り当てるには、ユーザー アクセス管理者ロール、または Microsoft.Authorization/roleAssignments/write アクションを含む別のロールが必要です。

Azure portal、Azure CLI、または Azure PowerShell を使用して、ユーザーに Azure RBAC ロールを割り当てることができます。 ロールの割り当てに使用できるスコープの詳細については、「 Azure RBAC のスコープについて」を参照してください。

次の例では、ユーザー アカウントに、Azure Service Bus Data Owner ロールを割り当てます。これにより、Azure Service Bus リソースへのフル アクセスが提供されます。 実際のシナリオでは、 最小限の特権の原則 に従って、より安全な運用環境に必要な最小限のアクセス許可のみをユーザーに付与します。

Azure Service Bus 用の Azure 組み込みロール

Azure Service Bus の場合、名前空間およびそれに関連するすべてのリソースの Azure portal および Azure リソース管理 API による管理は、Azure RBAC モデルを使って既に保護されています。 Azure には、Service Bus 名前空間へのアクセスを承認するための次の Azure 組み込みロールが用意されています。

• Azure Service Bus データ所有者: Service Bus 名前空間とそのエンティティ (キュー、トピック、サブスクリプション、フィルターなど) へのデータ アクセスを有効にします。 このロールのメンバーは、キューまたはトピックやサブスクリプションとの間でメッセージを送受信できます。
• Azure Service Bus データ送信者: このロールを使用して、 send に Service Bus 名前空間とそのエンティティへのアクセス権を付与します。
• Azure Service Bus データ レシーバー: このロールを使用して、Service Bus 名前空間とそのエンティティに receive アクセス権を付与します。

カスタム ロールを作成する場合は、Service Bus 操作に必要な権限に関するページを参照してください。

Microsoft Entra ユーザーを Azure Service Bus 所有者ロールに追加する

Microsoft Entra ユーザー名を、Service Bus 名前空間レベルの Azure Service Bus データ所有者ロール に追加します。 この構成により、ユーザー アカウントのコンテキストで実行されるアプリは、キューまたはトピックにメッセージを送信できます。 キューまたはトピックのサブスクリプションからメッセージを受信できます。

1. Azure portal で Service Bus 名前空間ページが開いていない場合は、メイン検索バーまたは左側のナビゲーションを使用して Service Bus 名前空間を見つけます。

2. [ 概要 ] ページで、左側のメニューから [アクセス制御 (IAM)] を選択します。

3. [アクセス制御 (IAM)] ページで、[ロールの割り当て] タブを選びます。

4. 上部のメニューから [ + 追加] を選択し、[ ロールの割り当ての追加] を選択します。

   

5. 検索ボックスを使って、結果を目的のロールに絞り込みます。 この例では、Azure Service Bus Data Owner を検索して一致する結果を選択します。 [次へ] を選びます。

6. [アクセスの割り当て先] で、[ユーザー、グループ、またはサービス プリンシパル] を選び、[+ メンバーの選択] を選びます。

7. ダイアログで、自分の Microsoft Entra ユーザー名 (通常は user@___domain メール アドレス) を検索し、ダイアログの下部にある [選択] を選びます。

8. [レビューと割り当て] を選んで最終ページに移動し、もう一度 [レビューと割り当て] を行ってプロセスを完了します。

接続文字列を取得する

新しい名前空間を作成すると、プライマリ キーとセカンダリ キーを含む初期 Shared Access Signature (SAS) ポリシーが自動的に生成されます。 プライマリ接続文字列とセカンダリ接続文字列が作成され、それぞれが名前空間のすべての側面を完全に制御できます。 通常の送信者と受信者に対してより制約のある権限を持つ規則を作成する方法の詳細については、 Service Bus の認証と承認に関するセクションを参照してください。

クライアントは、接続文字列を使用して Service Bus の名前空間に接続できます。 名前空間のプライマリ接続文字列をコピーするには、次の手順に従います。

1. [Service Bus 名前空間] ページの左側のメニューで、[設定] を展開し、[共有アクセス ポリシー] を選択します。

2. [共有アクセス ポリシー] ページで、 [RootManageSharedAccessKey] を選択します。

3. [ポリシー: RootManageSharedAccessKey] ウィンドウで、[プライマリ接続文字列] の横にある [コピー] ボタンを選択し、後で使用するために接続文字列をクリップボードにコピーします。 この値をメモ帳などに一時的に貼り付けます。

   

   このページを使用して、主キー、2 次キー、プライマリ接続文字列、セカンダリ接続文字列をコピーできます。

1. この Service Bus クイック スタートに必要な Python パッケージをインストールするには、パスに Python があるコマンド プロンプト ウィンドウを開きます。

2. サンプルを配置するフォルダーにディレクトリを変更します。

3. 次のパッケージをインストールします。

   pip install azure-servicebus
   pip install azure-identity
   pip install aiohttp
   

1. この Service Bus クイック スタートに必要な Python パッケージをインストールするには、パスに Python があるコマンド プロンプト ウィンドウを開きます。

2. サンプルを配置するフォルダーにディレクトリを変更します。

3. 次のパッケージをインストールします。

   pip install azure-servicebus
   

1. import ステートメントを追加します。

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

2. 定数を追加し、資格情報を定義します。

   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   QUEUE_NAME = "QUEUE_NAME"
   
   credential = DefaultAzureCredential()
   

   重要

   • FULLY_QUALIFIED_NAMESPACE を Service Bus 名前空間の完全修飾名前空間に置き換えます。
   • QUEUE_NAME をキューの名前に置き換えます。

3. 単一のメッセージを送信するためのメソッドを追加します。

   async def send_single_message(sender):
       # Create a Service Bus message and send it to the queue
       message = ServiceBusMessage("Single Message")
       await sender.send_messages(message)
       print("Sent a single message")
   

   sender は、作成したキューのクライアントとしての役割を担うオブジェクトです。 後で作成し、この関数に引数として送信します。

4. 一連のメッセージを送信するためのメソッドを追加します。

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. メッセージのバッチを送信するためのメソッドを追加します。

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Service Bus クライアントを作成し、メッセージを送信するためのキューの sender オブジェクトを作成します。

   async def run():
       # create a Service Bus client using the credential
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
           # get a Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
           async with sender:
               # send one message
               await send_single_message(sender)
               # send a list of messages
               await send_a_list_of_messages(sender)
               # send a batch of messages
               await send_batch_message(sender)
   
           # Close credential when no longer needed.
           await credential.close()
   

7. run メソッドを呼び出し、メッセージを出力します。

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. import ステートメントを追加します。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   from azure.servicebus import ServiceBusMessage
   

2. 定数を追加します。

   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

   重要

   • NAMESPACE_CONNECTION_STR を Service Bus 名前空間の接続文字列に置き換えます。
   • QUEUE_NAME をキューの名前に置き換えます。

3. 単一のメッセージを送信するためのメソッドを追加します。

   async def send_single_message(sender):
       # Create a Service Bus message and send it to the queue
       message = ServiceBusMessage("Single Message")
       await sender.send_messages(message)
       print("Sent a single message")
   

   sender は、作成したキューのクライアントとしての役割を担うオブジェクトです。 後で作成し、この関数に引数として送信します。

4. 一連のメッセージを送信するためのメソッドを追加します。

   async def send_a_list_of_messages(sender):
       # Create a list of messages and send it to the queue
       messages = [ServiceBusMessage("Message in list") for _ in range(5)]
       await sender.send_messages(messages)
       print("Sent a list of 5 messages")
   

5. メッセージのバッチを送信するためのメソッドを追加します。

   async def send_batch_message(sender):
       # Create a batch of messages
       async with sender:
           batch_message = await sender.create_message_batch()
           for _ in range(10):
               try:
                   # Add a message to the batch
                   batch_message.add_message(ServiceBusMessage("Message inside a ServiceBusMessageBatch"))
               except ValueError:
                   # ServiceBusMessageBatch object reaches max_size.
                   # New ServiceBusMessageBatch object can be created here to send more data.
                   break
           # Send the batch of messages to the queue
           await sender.send_messages(batch_message)
       print("Sent a batch of 10 messages")
   

6. Service Bus クライアントを作成し、メッセージを送信するためのキューの sender オブジェクトを作成します。

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
           # Get a Queue Sender object to send messages to the queue
           sender = servicebus_client.get_queue_sender(queue_name=QUEUE_NAME)
           async with sender:
               # Send one message
               await send_single_message(sender)
               # Send a list of messages
               await send_a_list_of_messages(sender)
               # Send a batch of messages
               await send_batch_message(sender)
   

7. run メソッドを呼び出し、メッセージを出力します。

   asyncio.run(run())
   print("Done sending messages")
   print("-----------------------")
   

1. send.py サンプルと同様に、importステートメントを追加します。 定数を独自の値に置き換え、資格情報を定義します。

   import asyncio
   
   from azure.servicebus.aio import ServiceBusClient
   from azure.identity.aio import DefaultAzureCredential
   
   FULLY_QUALIFIED_NAMESPACE = "FULLY_QUALIFIED_NAMESPACE"
   QUEUE_NAME = "QUEUE_NAME"
   
   credential = DefaultAzureCredential()
   

2. Service Bus クライアントを、次にメッセージを受信するためのキューの receiver オブジェクトを作成します。

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient(
           fully_qualified_namespace=FULLY_QUALIFIED_NAMESPACE,
           credential=credential,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the queue
                       await receiver.complete_message(msg)
   
           # Close credential when no longer needed.
           await credential.close()
   

3. run メソッドを呼び出します。

   asyncio.run(run())
   

1. send.py サンプルと同様に、importステートメントを追加し、独自の値を使用する定数を定義します。

   import asyncio
   from azure.servicebus.aio import ServiceBusClient
   
   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

2. Service Bus クライアントを、次にメッセージを受信するためのキューの receiver オブジェクトを作成します。

   async def run():
       # create a Service Bus client using the connection string
       async with ServiceBusClient.from_connection_string(
           conn_str=NAMESPACE_CONNECTION_STR,
           logging_enable=True) as servicebus_client:
   
           async with servicebus_client:
               # get the Queue Receiver object for the queue
               receiver = servicebus_client.get_queue_receiver(queue_name=QUEUE_NAME)
               async with receiver:
                   received_msgs = await receiver.receive_messages(max_wait_time=5, max_message_count=20)
                   for msg in received_msgs:
                       print("Received: " + str(msg))
                       # complete the message so that the message is removed from the queue
                       await receiver.complete_message(msg)
   

3. run メソッドを呼び出します。

   asyncio.run(run())