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