Azure Storage Queue Py

作者 microsoft354361d83247MIT收录于 2026年10月8日更新于 2026年10月8日

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

精选仅含说明Software Development DevOps & Cloud

AI 生成的概览

指导 Python 开发者使用 Azure 队列存储 SDK 进行消息队列、任务分发和异步处理。

功能: 该技能提供 Azure 队列存储 Python SDK 的参考文档和代码示例。内容涵盖安装、使用 DefaultAzureCredential 进行身份验证、队列生命周期操作、发送和接收消息、查看、更新、删除、清空队列以及异步客户端用法。还包括可靠消息处理和二进制数据编码的最佳实践。
适用场景: 当构建需要可靠消息队列、任务分发或使用 Azure 队列存储进行异步处理的 Python 应用时使用此技能。适用于实现基于队列的工作流、后台作业处理或解耦服务通信的开发者。
运行要求: 需要安装 azure-storage-queue 和 azure-identity 包的 Python 环境。需要 Azure 存储账户 URL 和适当的凭据，通常通过 DefaultAzureCredential 提供。需要访问 Azure 服务的网络连接。该技能不包含脚本，仅为说明和参考文档。

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.

来源与署名

来源：microsoft/skills位于.github/plugins/azure-sdk-python/skills/azure-storage-queue-py提交354361d

许可证： MIT

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架

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

来源与署名

更多来自 microsoft/skills 的技能

Customize

Discover Azure Skills

Azure Resource Visualizer

Azure Resource Lookup

Azure Messaging

Azure Kusto

更多Software Development技能

Playground

M5 Onboard

Cardputer Buddy

Testing Strategy

Tech Debt

System Design