Send messages to and receive messages from Azure Service Bus queues (Python)

To use this quickstart with your own Azure account:

• Install Azure CLI, which provides the passwordless authentication to your developer machine.
• Set Azure Environment as Azure operated by 21Vianet with az cloud set -n AzureChinaCloud.
• Sign in with your Azure account at the terminal or command prompt with az login.
• Use the same account when you add the appropriate data role to your resource.
• Run the code in the same terminal or command prompt.
• Note the queue name for your Service Bus namespace. You'll need that in the code.

Note the following, which you'll use in the code below:

• Service Bus namespace connection string
• Service Bus namespace queue you created

Assign roles to your Azure AD user

When developing locally, make sure that the user account that connects to Azure Service Bus has the correct permissions. You'll need the Azure Service Bus Data Owner role in order to send and receive messages. To assign yourself this role, you'll need the User Access Administrator role, or another role that includes the Microsoft.Authorization/roleAssignments/write action. You can assign Azure RBAC roles to a user using the Azure portal, Azure CLI, or Azure PowerShell. Learn more about the available scopes for role assignments on the scope overview page.

The following example assigns the Azure Service Bus Data Owner role to your user account, which provides full access to Azure Service Bus resources. In a real scenario, follow the Principle of Least Privilege to give users only the minimum permissions needed for a more secure production environment.

Azure built-in roles for Azure Service Bus

For Azure Service Bus, the management of namespaces and all related resources through the Azure portal and the Azure resource management API is already protected using the Azure RBAC model. Azure provides the below Azure built-in roles for authorizing access to a Service Bus namespace:

• Azure Service Bus Data Owner: Enables data access to Service Bus namespace and its entities (queues, topics, subscriptions, and filters). A member of this role can send and receive messages from queues or topics/subscriptions.
• Azure Service Bus Data Sender: Use this role to give the send access to Service Bus namespace and its entities.
• Azure Service Bus Data Receiver: Use this role to give the receive access to Service Bus namespace and its entities.

If you want to create a custom role, see Rights required for Service Bus operations.

Add Azure AD user to Azure Service Bus Owner role

Add your Azure AD user name to the Azure Service Bus Data Owner role at the Service Bus namespace level. It will allow an app running in the context of your user account to send messages to a queue or a topic, and receive messages from a queue or a topic's subscription.

1. If you don't have the Service Bus Namespace page open in the Azure portal, locate your Service Bus namespace using the main search bar or left navigation.

2. On the overview page, select Access control (IAM) from the left-hand menu.

3. On the Access control (IAM) page, select the Role assignments tab.

4. Select + Add from the top menu and then Add role assignment from the resulting drop-down menu.

   

5. Use the search box to filter the results to the desired role. For this example, search for Azure Service Bus Data Owner and select the matching result. Then choose Next.

6. Under Assign access to, select User, group, or service principal, and then choose + Select members.

7. In the dialog, search for your Azure AD username (usually your user@domain email address) and then choose Select at the bottom of the dialog.

8. Select Review + assign to go to the final page, and then Review + assign again to complete the process.

Get the connection string

Creating a new namespace automatically generates an initial Shared Access Signature (SAS) policy with primary and secondary keys, and primary and secondary connection strings that each grant full control over all aspects of the namespace. See Service Bus authentication and authorization for information about how to create rules with more constrained rights for regular senders and receivers.

A client can use the connection string to connect to the Service Bus namespace. To copy the primary connection string for your namespace, follow these steps:

1. On the Service Bus Namespace page, select Shared access policies on the left menu.

2. On the Shared access policies page, select RootManageSharedAccessKey.

3. In the Policy: RootManageSharedAccessKey window, select the copy button next to Primary Connection String, to copy the connection string to your clipboard for later use. Paste this value into Notepad or some other temporary location.

   

   You can use this page to copy primary key, secondary key, primary connection string, and secondary connection string.

1. To install the required Python packages for this Service Bus tutorial, open a command prompt that has Python in its path, change the directory to the folder where you want to have your samples.

2. Install the following packages:

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

1. To install the required Python packages for this Service Bus tutorial, open a command prompt that has Python in its path, change the directory to the folder where you want to have your samples.

2. Install the following package:

   pip install azure-servicebus
   

1. Add import statements.

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

2. Add constants and define a credential.

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

   Important

   • Replace FULLY_QUALIFIED_NAMESPACE with the fully qualified namespace for your Service Bus namespace.
   • Replace QUEUE_NAME with the name of the queue.

3. Add a method to send a single message.

   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")
   

   The sender is an object that acts as a client for the queue you created. You'll create it later and send as an argument to this function.

4. Add a method to send a list of messages.

   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. Add a method to send a batch of messages.

   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. Create a Service Bus client and then a queue sender object to send messages.

   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. Call the run method and print a message.

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

1. Add import statements.

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

2. Add constants.

   NAMESPACE_CONNECTION_STR = "NAMESPACE_CONNECTION_STR"
   QUEUE_NAME = "QUEUE_NAME"
   

   Important

   • Replace NAMESPACE_CONNECTION_STR with the connection string for your Service Bus namespace.
   • Replace QUEUE_NAME with the name of the queue.

3. Add a method to send a single message.

   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")
   

   The sender is an object that acts as a client for the queue you created. You'll create it later and send as an argument to this function.

4. Add a method to send a list of messages.

   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. Add a method to send a batch of messages.

   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. Create a Service Bus client and then a queue sender object to send messages.

   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. Call the run method and print a message.

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

1. Similar to the send sample, add import statements, define constants that you should replace with your own values, and define a credential.

   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. Create a Service Bus client and then a queue receiver object to receive messages.

   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. Call the run method.

   asyncio.run(run())
   

1. Similar to the send sample, add import statements and define constants that you should replace with your own values.

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

2. Create a Service Bus client and then a queue receiver object to receive messages.

   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. Call the run method.

   asyncio.run(run())