azure-servicebus-py
microsoft/skills
Python SDK を使用して、Azure Service Bus のキュー、トピック、サブスクリプションでメッセージを送受信します。
...すべて拡張しますAzure Service Bus Python SDK
キューおよびパブリッシュ/サブスクライブ(pub/sub)トピックを使用した、信頼性の高いクラウド通信のためのエンタープライズ向けメッセージング。
インストール
pip install azure-servicebus azure-identity
環境変数
SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net # すべての認証方法で必須
SERVICEBUS_QUEUE_NAME=myqueue # キュー操作で必須
SERVICEBUS_TOPIC_NAME=mytopic # トピック操作で必須
SERVICEBUS_SUBSCRIPTION_NAME=mysubscription # サブスクリプション操作で必須
AZURE_TOKEN_CREDENTIALS=prod # 本番環境で DefaultAzureCredential を使用する場合にのみ必須
</namespace>認証とライフサイクル
🔑 以下のすべてのコードサンプルに適用される2つのルールがあります:
DefaultAzureCredentialを優先してください。 これにより、ローカル環境(Azure CLI / VS Code / Developer CLI)および Azure(マネージド ID、ワークロード ID)で、コードの変更なしに動作します。接続文字列やアカウント/API キーの使用は避けてください。これらは Entra の監査およびローテーションをバイパスするためです。
- ローカル開発:
DefaultAzureCredentialはそのまま使用できます。- 本番環境:
AZURE_TOKEN_CREDENTIALS=prod(またはAZURE_TOKEN_CREDENTIALS=<specific_credential></specific_credential>)を設定し、認証チェーンを本番環境で安全な資格情報に制限します。- すべてのクライアントをコンテキストマネージャーでラップしてください。 これにより、HTTP トランスポート、ソケット、トークンキャッシュが確定的に解放されます:
- 同期:
with <client>(...) as client:</client>- 非同期:
async with <client>(...) as client:</client>およびasync with DefaultAzureCredential() as credential:(azure.identity.aioから)スニペットではこの設定を省略している場合がありますが、本番環境のコードは常に両方のルールに従う必要があります。
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.servicebus import ServiceBusClient
# ローカル開発:DefaultAzureCredential。本番環境:AZURE_TOKEN_CREDENTIALS=prod または AZURE_TOKEN_CREDENTIALS=<specific_credential> を設定
credential = DefaultAzureCredential(require_envvar=True)
# または、本番環境で特定の資格情報を直接使用することもできます:
# 詳細は https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes を参照
# credential = ManagedIdentityCredential()
namespace = "<namespace>.servicebus.windows.net"
with ServiceBusClient(
fully_qualified_namespace=namespace,
credential=credential
) as client:
# ここからクライアントを使用(操作の詳細は以下のセクションを参照)
...
</namespace></specific_credential>クライアントの種類
| クライアント | 目的 | 取得方法 |
|---|---|---|
| `ServiceBusClient` | 接続管理 | 直接インスタンス化 |
| `ServiceBusSender` | メッセージの送信 | `client.get_queue_sender()` / `get_topic_sender()` |
| `ServiceBusReceiver` | メッセージの受信 | `client.get_queue_receiver()` / `get_subscription_receiver()` |
メッセージの送信(非同期)
import asyncio
from azure.servicebus.aio import ServiceBusClient
from azure.servicebus import ServiceBusMessage
from azure.identity.aio import DefaultAzureCredential
async def send_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
sender = client.get_queue_sender(queue_name="myqueue")
async with sender:
# 単一メッセージ
message = ServiceBusMessage("Hello, Service Bus!")
await sender.send_messages(message)
# メッセージのバッチ
messages = [ServiceBusMessage(f"Message {i}") for i in range(10)]
await sender.send_messages(messages)
# メッセージバッチ(サイズ制御用)
batch = await sender.create_message_batch()
for i in range(100):
try:
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
except ValueError: # バッチがいっぱいの場合
await sender.send_messages(batch)
batch = await sender.create_message_batch()
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
await sender.send_messages(batch)
asyncio.run(send_messages())
</namespace>メッセージの受信(非同期)
async def receive_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
receiver = client.get_queue_receiver(queue_name="myqueue")
async with receiver:
# バッチの受信
messages = await receiver.receive_messages(
max_message_count=10,
max_wait_time=5 # 秒
)
for msg in messages:
print(f"Received: {str(msg)}")
await receiver.complete_message(msg) # キューから削除
asyncio.run(receive_messages())
</namespace>受信モード
| モード | 動作 | 使用ケース |
|---|---|---|
| `PEEK_LOCK`(デフォルト) | メッセージがロックされ、完了または放棄が必要 | 信頼性の高い処理 |
| `RECEIVE_AND_DELETE` | 受信時に即座に削除 | 最大1回の配信 |
from azure.servicebus import ServiceBusReceiveMode
receiver = client.get_queue_receiver(
queue_name="myqueue",
receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE
)
メッセージの処理(セットルメント)
async with receiver:
messages = await receiver.receive_messages(max_message_count=1)
for msg in messages:
try:
# メッセージの処理...
await receiver.complete_message(msg) # 成功 - キューから削除
except ProcessingError:
await receiver.abandon_message(msg) # 後で再試行
except PermanentError:
await receiver.dead_letter_message(
msg,
reason="ProcessingFailed",
error_description="Could not process"
)
| アクション | 効果 |
|---|---|
| `complete_message()` | キューから削除(成功) |
| `abandon_message()` | ロックを解放し、直ちに再試行 |
| `dead_letter_message()` | デッドレターキューに移動 |
| `defer_message()` | 一時保管し、シーケンス番号で受信 |
トピックおよびサブスクリプション
# トピックへの送信
sender = client.get_topic_sender(topic_name="mytopic")
async with sender:
await sender.send_messages(ServiceBusMessage("Topic message"))
# サブスクリプションからの受信
receiver = client.get_subscription_receiver(
topic_name="mytopic",
subscription_name="mysubscription"
)
async with receiver:
messages = await receiver.receive_messages(max_message_count=10)
セッション(FIFO)
# セッション付きの送信
message = ServiceBusMessage("Session message")
message.session_id = "order-123"
await sender.send_messages(message)
# 特定のセッションからの受信
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id="order-123"
)
# 利用可能な次のセッションからの受信
from azure.servicebus import NEXT_AVAILABLE_SESSION
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id=NEXT_AVAILABLE_SESSION
)
スケジュールされたメッセージ
from datetime import datetime, timedelta, timezone
message = ServiceBusMessage("Scheduled message")
scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=10)
# メッセージのスケジュール
sequence_number = await sender.schedule_messages(message, scheduled_time)
# スケジュールされたメッセージのキャンセル
await sender.cancel_scheduled_messages(sequence_number)
デッドレターキュー
from azure.servicebus import ServiceBusSubQueue
# デッドレターキューからの受信
dlq_receiver = client.get_queue_receiver(
queue_name="myqueue",
sub_queue=ServiceBusSubQueue.DEAD_LETTER
)
async with dlq_receiver:
messages = await dlq_receiver.receive_messages(max_message_count=10)
for msg in messages:
print(f"Dead-lettered: {msg.dead_letter_reason}")
await dlq_receiver.complete_message(msg)
同期クライアント(単純なスクリプト用)
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.identity import DefaultAzureCredential
with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=DefaultAzureCredential()
) as client:
with client.get_queue_sender("myqueue") as sender:
sender.send_messages(ServiceBusMessage("Sync message"))
with client.get_queue_receiver("myqueue") as receiver:
for msg in receiver:
print(str(msg))
receiver.complete_message(msg)
</namespace>ベストプラクティス
- 同期または非同期のいずれかを選択し、一貫性を保ってください。 同一の呼び出しパス内で
azure.xxxの同期クライアントとazure.xxx.aioの非同期クライアントを混在させないでください。モジュールごとに1つのモードを選択してください。 - クライアントおよび非同期資格情報には常にコンテキストマネージャーを使用してください。 適切なクリーンアップのため、すべてのクライアントを
with Client(...) as client:(同期)またはasync with Client(...) as client:(非同期)でラップしてください。azure.identity.aioからの非同期DefaultAzureCredentialの場合、トークンおよびトランスポートのクリーンアップを行うため、async with credential:も使用してください。 - ポータブルな認証のために
DefaultAzureCredentialを使用してください。(ローカル開発および Azure 間で、可能な限り接続文字列や API キーを避けてください)。 - 本番環境のワークロードには非同期クライアントを使用してください。
- 処理が成功した後にメッセージを完了(complete)してください。
- ポイズンメッセージにはデッドレターキューを使用してください。
- 順序付きの FIFO 処理にはセッションを使用してください。
- 高スループットのシナリオにはメッセージバッチを使用してください。
- 無限のブロックを避けるため、
max_wait_timeを設定してください。
参照ファイル
| ファイル | 内容 |
|---|---|
| references/patterns.md | 競合する消費者、セッション、リトライパターン、リクエスト-レスポンス、トランザクション |
| references/dead-letter.md | DLQ 処理、ポイズンメッセージ、再処理戦略 |
| scripts/setup\_servicebus.py | キュー/トピック/サブスクリプション管理および DLQ モニタリングの CLI |
---
name: azure-servicebus-py
description: Send and receive messages using Azure Service Bus queues, topics, and subscriptions with Python SDK.
license: MIT
---
# Azure Service Bus SDK for Python
Enterprise messaging for reliable cloud communication with queues and pub/sub topics.
## Installation
```bash
pip install azure-servicebus azure-identity
```
## Environment Variables
```bash
SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net # Required for all auth methods
SERVICEBUS_QUEUE_NAME=myqueue # Required for queue operations
SERVICEBUS_TOPIC_NAME=mytopic # Required for topic operations
SERVICEBUS_SUBSCRIPTION_NAME=mysubscription # Required for subscription operations
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
```
## Authentication & Lifecycle
> **🔑 Two rules apply to every code sample below:**
>
> 1. **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.
> 2. **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, ManagedIdentityCredential
from azure.servicebus import ServiceBusClient
# 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()
namespace = "<namespace>.servicebus.windows.net"
with ServiceBusClient(
fully_qualified_namespace=namespace,
credential=credential
) as client:
# Use client here (see following sections for operations)
...
```
## Client Types
| Client | Purpose | Get From |
|--------|---------|----------|
| `ServiceBusClient` | Connection management | Direct instantiation |
| `ServiceBusSender` | Send messages | `client.get_queue_sender()` / `get_topic_sender()` |
| `ServiceBusReceiver` | Receive messages | `client.get_queue_receiver()` / `get_subscription_receiver()` |
## Send Messages (Async)
```python
import asyncio
from azure.servicebus.aio import ServiceBusClient
from azure.servicebus import ServiceBusMessage
from azure.identity.aio import DefaultAzureCredential
async def send_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
sender = client.get_queue_sender(queue_name="myqueue")
async with sender:
# Single message
message = ServiceBusMessage("Hello, Service Bus!")
await sender.send_messages(message)
# Batch of messages
messages = [ServiceBusMessage(f"Message {i}") for i in range(10)]
await sender.send_messages(messages)
# Message batch (for size control)
batch = await sender.create_message_batch()
for i in range(100):
try:
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
except ValueError: # Batch full
await sender.send_messages(batch)
batch = await sender.create_message_batch()
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
await sender.send_messages(batch)
asyncio.run(send_messages())
```
## Receive Messages (Async)
```python
async def receive_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
receiver = client.get_queue_receiver(queue_name="myqueue")
async with receiver:
# Receive batch
messages = await receiver.receive_messages(
max_message_count=10,
max_wait_time=5 # seconds
)
for msg in messages:
print(f"Received: {str(msg)}")
await receiver.complete_message(msg) # Remove from queue
asyncio.run(receive_messages())
```
## Receive Modes
| Mode | Behavior | Use Case |
|------|----------|----------|
| `PEEK_LOCK` (default) | Message locked, must complete/abandon | Reliable processing |
| `RECEIVE_AND_DELETE` | Removed immediately on receive | At-most-once delivery |
```python
from azure.servicebus import ServiceBusReceiveMode
receiver = client.get_queue_receiver(
queue_name="myqueue",
receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE
)
```
## Message Settlement
```python
async with receiver:
messages = await receiver.receive_messages(max_message_count=1)
for msg in messages:
try:
# Process message...
await receiver.complete_message(msg) # Success - remove from queue
except ProcessingError:
await receiver.abandon_message(msg) # Retry later
except PermanentError:
await receiver.dead_letter_message(
msg,
reason="ProcessingFailed",
error_description="Could not process"
)
```
| Action | Effect |
|--------|--------|
| `complete_message()` | Remove from queue (success) |
| `abandon_message()` | Release lock, retry immediately |
| `dead_letter_message()` | Move to dead-letter queue |
| `defer_message()` | Set aside, receive by sequence number |
## Topics and Subscriptions
```python
# Send to topic
sender = client.get_topic_sender(topic_name="mytopic")
async with sender:
await sender.send_messages(ServiceBusMessage("Topic message"))
# Receive from subscription
receiver = client.get_subscription_receiver(
topic_name="mytopic",
subscription_name="mysubscription"
)
async with receiver:
messages = await receiver.receive_messages(max_message_count=10)
```
## Sessions (FIFO)
```python
# Send with session
message = ServiceBusMessage("Session message")
message.session_id = "order-123"
await sender.send_messages(message)
# Receive from specific session
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id="order-123"
)
# Receive from next available session
from azure.servicebus import NEXT_AVAILABLE_SESSION
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id=NEXT_AVAILABLE_SESSION
)
```
## Scheduled Messages
```python
from datetime import datetime, timedelta, timezone
message = ServiceBusMessage("Scheduled message")
scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=10)
# Schedule message
sequence_number = await sender.schedule_messages(message, scheduled_time)
# Cancel scheduled message
await sender.cancel_scheduled_messages(sequence_number)
```
## Dead-Letter Queue
```python
from azure.servicebus import ServiceBusSubQueue
# Receive from dead-letter queue
dlq_receiver = client.get_queue_receiver(
queue_name="myqueue",
sub_queue=ServiceBusSubQueue.DEAD_LETTER
)
async with dlq_receiver:
messages = await dlq_receiver.receive_messages(max_message_count=10)
for msg in messages:
print(f"Dead-lettered: {msg.dead_letter_reason}")
await dlq_receiver.complete_message(msg)
```
## Sync Client (for simple scripts)
```python
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.identity import DefaultAzureCredential
with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=DefaultAzureCredential()
) as client:
with client.get_queue_sender("myqueue") as sender:
sender.send_messages(ServiceBusMessage("Sync message"))
with client.get_queue_receiver("myqueue") as receiver:
for msg in receiver:
print(str(msg))
receiver.complete_message(msg)
```
## Best Practices
1. **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.
2. **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 proper cleanup. For async `DefaultAzureCredential` from `azure.identity.aio`, also use `async with credential:` so tokens and transports are cleaned up.
3. **Use `DefaultAzureCredential`** for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
4. **Use async client** for production workloads
5. **Complete messages** after successful processing
6. **Use dead-letter queue** for poison messages
7. **Use sessions** for ordered, FIFO processing
8. **Use message batches** for high-throughput scenarios
9. **Set `max_wait_time`** to avoid infinite blocking
## Reference Files
| File | Contents |
|------|----------|
| [references/patterns.md](references/patterns.md) | Competing consumers, sessions, retry patterns, request-response, transactions |
| [references/dead-letter.md](references/dead-letter.md) | DLQ handling, poison messages, reprocessing strategies |
| [scripts/setup_servicebus.py](scripts/setup_servicebus.py) | CLI for queue/topic/subscription management and DLQ monitoring |
すべてのファイル
0件のファイルazure-servicebus-pyをインストール
スキルファイルをダウンロードして .claude/skills/ ディレクトリに展開してください。
ZIPをダウンロードリポジトリをクローンし、スキルファイルをプロジェクトにコピーしてください。
git clone https://github.com/microsoft/skills/tree/main/.github/plugins/azure-sdk-python/skills/azure-servicebus-py # Copy SKILL.md to your .claude/skills/ directory
コピー





家
