Azure Storage Queue Py

by microsoft354361d83247MITListed Oct 8, 2026Updated Oct 8, 2026

Azure Queue Storage SDK for Python. Use for reliable message queuing, task distribution, and asynchronous processing. Triggers: "queue storage", "QueueServiceClient", "QueueClient", "message queue", "dequeue".

FeaturedInstructions onlySoftware Development DevOps & Cloud

AI-generated overview

Guides Python developers in using the Azure Queue Storage SDK for message queuing, task distribution, and asynchronous processing.

What it does: This skill provides reference documentation and code examples for the Azure Queue Storage SDK for Python. It covers installation, authentication with DefaultAzureCredential, queue lifecycle operations, sending and receiving messages, peeking, updating, deleting, clearing queues, and async client usage. It also includes best practices for reliable message processing and binary data encoding.
When to use it: Use this skill when building Python applications that need reliable message queuing, task distribution, or asynchronous processing with Azure Queue Storage. It is suited for developers implementing queue-based workflows, background job processing, or decoupled service communication.
Requirements: Requires Python with the azure-storage-queue and azure-identity packages installed. Needs an Azure Storage account URL and appropriate credentials, typically via DefaultAzureCredential. Network access to Azure services is required. The skill ships no scripts; it is instructions and reference documentation only.

Azure Queue Storage SDK for Python

Simple, cost-effective message queuing for asynchronous communication.

Installation

Environment Variables

bash

AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net  # Required for all auth methodsAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

AZURE_STORAGE_ACCOUNT_URL=https://<account>.queue.core.windows.net  # Required for all auth methodsAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

Authentication & Lifecycle

🔑 Two rules apply to every code sample below:

Prefer DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.

Local dev: DefaultAzureCredential works as-is.

Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.

Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:

Sync: with <Client>(...) as client:

Async: async with <Client>(...) as client: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)

Snippets may abbreviate this setup, but production code should always follow both rules.

python

from azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.storage.queue import QueueServiceClient, QueueClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>credential = DefaultAzureCredential(require_envvar=True)# Or use a specific credential directly in production:# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes# credential = ManagedIdentityCredential()account_url = "https://<account>.queue.core.windows.net"
# Service clientwith QueueServiceClient(account_url=account_url, credential=credential) as service_client:    # Use service_client here (see following sections for operations)    ...
# Queue clientwith QueueClient(account_url=account_url, queue_name="myqueue", credential=credential) as queue_client:    # Use queue_client here (see following sections for operations)    ...

from azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.storage.queue import QueueServiceClient, QueueClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>credential = DefaultAzureCredential(require_envvar=True)# Or use a specific credential directly in production:# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes# credential = ManagedIdentityCredential()account_url = "https://<account>.queue.core.windows.net"
# Service clientwith QueueServiceClient(account_url=account_url, credential=credential) as service_client:    # Use service_client here (see following sections for operations)    ...
# Queue clientwith QueueClient(account_url=account_url, queue_name="myqueue", credential=credential) as queue_client:    # Use queue_client here (see following sections for operations)    ...

Queue Operations

python

# Create queueservice_client.create_queue("myqueue")
# Get queue clientqueue_client = service_client.get_queue_client("myqueue")
# Delete queueservice_client.delete_queue("myqueue")
# List queuesfor queue in service_client.list_queues():    print(queue.name)

# Create queueservice_client.create_queue("myqueue")
# Get queue clientqueue_client = service_client.get_queue_client("myqueue")
# Delete queueservice_client.delete_queue("myqueue")
# List queuesfor queue in service_client.list_queues():    print(queue.name)

Send Messages

python

# Send message (string)queue_client.send_message("Hello, Queue!")
# Send with optionsqueue_client.send_message(    content="Delayed message",    visibility_timeout=60,  # Hidden for 60 seconds    time_to_live=3600       # Expires in 1 hour)
# Send JSONimport jsondata = {"task": "process", "id": 123}queue_client.send_message(json.dumps(data))

# Send message (string)queue_client.send_message("Hello, Queue!")
# Send with optionsqueue_client.send_message(    content="Delayed message",    visibility_timeout=60,  # Hidden for 60 seconds    time_to_live=3600       # Expires in 1 hour)
# Send JSONimport jsondata = {"task": "process", "id": 123}queue_client.send_message(json.dumps(data))

Receive Messages

python

# Receive messages (makes them invisible temporarily)messages = queue_client.receive_messages(    messages_per_page=10,    visibility_timeout=30  # 30 seconds to process)
for message in messages:    print(f"ID: {message.id}")    print(f"Content: {message.content}")    print(f"Dequeue count: {message.dequeue_count}")        # Process message...        # Delete after processing    queue_client.delete_message(message)

# Receive messages (makes them invisible temporarily)messages = queue_client.receive_messages(    messages_per_page=10,    visibility_timeout=30  # 30 seconds to process)
for message in messages:    print(f"ID: {message.id}")    print(f"Content: {message.content}")    print(f"Dequeue count: {message.dequeue_count}")        # Process message...        # Delete after processing    queue_client.delete_message(message)

Peek Messages

python

# Peek without hiding (doesn't affect visibility)messages = queue_client.peek_messages(max_messages=5)
for message in messages:    print(message.content)

# Peek without hiding (doesn't affect visibility)messages = queue_client.peek_messages(max_messages=5)
for message in messages:    print(message.content)

Update Message

python

# Extend visibility or update contentmessages = queue_client.receive_messages()for message in messages:    # Extend timeout (need more time)    queue_client.update_message(        message,        visibility_timeout=60    )        # Update content and timeout    queue_client.update_message(        message,        content="Updated content",        visibility_timeout=60    )

# Extend visibility or update contentmessages = queue_client.receive_messages()for message in messages:    # Extend timeout (need more time)    queue_client.update_message(        message,        visibility_timeout=60    )        # Update content and timeout    queue_client.update_message(        message,        content="Updated content",        visibility_timeout=60    )

Delete Message

python

# Delete after successful processingmessages = queue_client.receive_messages()for message in messages:    try:        # Process...        queue_client.delete_message(message)    except Exception:        # Message becomes visible again after timeout        pass

# Delete after successful processingmessages = queue_client.receive_messages()for message in messages:    try:        # Process...        queue_client.delete_message(message)    except Exception:        # Message becomes visible again after timeout        pass

Clear Queue

Queue Properties

python

# Get queue propertiesproperties = queue_client.get_queue_properties()print(f"Approximate message count: {properties.approximate_message_count}")
# Set/get metadataqueue_client.set_queue_metadata(metadata={"environment": "production"})properties = queue_client.get_queue_properties()print(properties.metadata)

# Get queue propertiesproperties = queue_client.get_queue_properties()print(f"Approximate message count: {properties.approximate_message_count}")
# Set/get metadataqueue_client.set_queue_metadata(metadata={"environment": "production"})properties = queue_client.get_queue_properties()print(properties.metadata)

Async Client

python

from azure.storage.queue.aio import QueueServiceClient, QueueClientfrom azure.identity.aio import DefaultAzureCredential
async def queue_operations():    credential = DefaultAzureCredential()        async with QueueClient(        account_url="https://<account>.queue.core.windows.net",        queue_name="myqueue",        credential=credential    ) as client:        # Send        await client.send_message("Async message")                # Receive        async for message in client.receive_messages():            print(message.content)            await client.delete_message(message)
import asyncioasyncio.run(queue_operations())

from azure.storage.queue.aio import QueueServiceClient, QueueClientfrom azure.identity.aio import DefaultAzureCredential
async def queue_operations():    credential = DefaultAzureCredential()        async with QueueClient(        account_url="https://<account>.queue.core.windows.net",        queue_name="myqueue",        credential=credential    ) as client:        # Send        await client.send_message("Async message")                # Receive        async for message in client.receive_messages():            print(message.content)            await client.delete_message(message)
import asyncioasyncio.run(queue_operations())

Base64 Encoding

python

from azure.storage.queue import QueueClient, BinaryBase64EncodePolicy, BinaryBase64DecodePolicy
# For binary datawith QueueClient(    account_url=account_url,    queue_name="myqueue",    credential=credential,    message_encode_policy=BinaryBase64EncodePolicy(),    message_decode_policy=BinaryBase64DecodePolicy()) as queue_client:    # Send bytes    queue_client.send_message(b"Binary content")

from azure.storage.queue import QueueClient, BinaryBase64EncodePolicy, BinaryBase64DecodePolicy
# For binary datawith QueueClient(    account_url=account_url,    queue_name="myqueue",    credential=credential,    message_encode_policy=BinaryBase64EncodePolicy(),    message_decode_policy=BinaryBase64DecodePolicy()) as queue_client:    # Send bytes    queue_client.send_message(b"Binary content")

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async clients in the same call path. Choose one mode per module.
Always use context managers for clients and async credentials. Wrap every client in with Client(...) as client: (sync) or async with Client(...) as client: (async). For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
Delete messages after processing to prevent reprocessing
Set appropriate visibility timeout based on processing time
Handle dequeue_count for poison message detection
Use async client for high-throughput scenarios
Use peek_messages for monitoring without affecting queue
Set time_to_live to prevent stale messages
Consider Service Bus for advanced features (sessions, topics)

Reference Files

File	Contents
references/capabilities.md [blocked]	Additional non-hero capabilities, operation-group coverage, and production checklists.
references/non-hero-scenarios.md [blocked]	Dedicated non-hero examples for secondary/advanced scenarios.

Source and attribution

Source:microsoft/skillsin.github/plugins/azure-sdk-python/skills/azure-storage-queue-pyat commit354361d

License: MIT

Content belongs to its original authors. SourceWeft indexes it from a public repository.

Report or request removal

Azure Storage Queue Py

Azure Queue Storage SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Queue Operations

Send Messages

Receive Messages

Peek Messages

Update Message

Delete Message

Clear Queue

Queue Properties

Async Client

Base64 Encoding

Best Practices

Reference Files

Source and attribution

More from microsoft/skills

Customize

Discover Azure Skills

Azure Resource Visualizer

Azure Resource Lookup

Azure Messaging

Azure Kusto

More in Software Development

Playground

M5 Onboard

Cardputer Buddy

Dev

Dpop Adoption

Web Design Guidelines