azure-storage-queue

import os, uuid from azure.identity import DefaultAzureCredential from azure.storage.queue import QueueServiceClient, QueueClient, BinaryBase64EncodePolicy, BinaryBase64DecodePolicy try: print('Azure Queue storage - Python quickstart sample') # Retrieve the connection string for use with the application. # The storage connection string is a key for accessing your storage account. # It is recommended to use passwordless authentication in production. connect_str = os.environ.get('AZURE_STORAGE_CONNECTION_STRING', '') queue_name = "quickstart-" + str(uuid.uuid4()) # Create the QueueServiceClient object if connect_str: # Authenticate with connection string queue_service_client = QueueServiceClient.from_connection_string( connect_str, message_encode_policy=BinaryBase64EncodePolicy(), message_decode_policy=BinaryBase64DecodePolicy() ) print("Using connection string for authentication.") else: # Authenticate with DefaultAzureCredential account_url = os.environ.get('AZURE_STORAGE_ACCOUNT_URL') # e.g., 'https://<account-name>.queue.core.windows.net' if not account_url: raise ValueError("Please set AZURE_STORAGE_CONNECTION_STRING or AZURE_STORAGE_ACCOUNT_URL environment variable.") credential = DefaultAzureCredential() queue_service_client = QueueServiceClient( account_url=account_url, credential=credential, message_encode_policy=BinaryBase64EncodePolicy(), message_decode_policy=BinaryBase64DecodePolicy() ) print("Using DefaultAzureCredential for authentication.") # Get a client to interact with the queue queue_client = queue_service_client.get_queue_client(queue_name) # Create a queue print(f"Creating queue: {queue_name}") queue_client.create_queue() # Send a message message1 = "Hello, Azure Queue!" print(f"Adding message: {message1}") queue_client.send_message(message1) message2 = "This is a second message." print(f"Adding message: {message2}") queue_client.send_message(message2) # Peek at messages print("Peeking at messages...") peeked_messages = queue_client.peek_messages(max_messages=5) for peeked_message in peeked_messages: print(f"Peeked message: {peeked_message.content}") # Receive and delete messages print("Receiving and deleting messages...") messages = queue_client.receive_messages(messages_per_page=1) for message in messages: print(f"Received message: {message.content}") queue_client.delete_message(message) print(f"Deleted message: {message.content}") print(f"Successfully completed the quickstart. Deleting queue: {queue_name}") queue_client.delete_queue() except Exception as ex: print(f"Exception: {ex}")