Azure Servicebus Py

1---
2name: azure-servicebus-py
3description: |
4  Azure Service Bus SDK for Python messaging. Use for queues, topics, subscriptions, and enterprise messaging patterns.
5  Triggers: "service bus", "ServiceBusClient", "queue", "topic", "subscription", "message broker".
6package: azure-servicebus
7---
8 
9# Azure Service Bus SDK for Python
10 
11Enterprise messaging for reliable cloud communication with queues and pub/sub topics.
12 
13## Installation
14 
15```bash
16pip install azure-servicebus azure-identity
17```
18 
19## Environment Variables
20 
21```bash
22SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net
23SERVICEBUS_QUEUE_NAME=myqueue
24SERVICEBUS_TOPIC_NAME=mytopic
25SERVICEBUS_SUBSCRIPTION_NAME=mysubscription
26```
27 
28## Authentication
29 
30```python
31from azure.identity import DefaultAzureCredential
32from azure.servicebus import ServiceBusClient
33 
34credential = DefaultAzureCredential()
35namespace = "<namespace>.servicebus.windows.net"
36 
37client = ServiceBusClient(
38    fully_qualified_namespace=namespace,
39    credential=credential
40)
41```
42 
43## Client Types
44 
45| Client | Purpose | Get From |
46|--------|---------|----------|
47| `ServiceBusClient` | Connection management | Direct instantiation |
48| `ServiceBusSender` | Send messages | `client.get_queue_sender()` / `get_topic_sender()` |
49| `ServiceBusReceiver` | Receive messages | `client.get_queue_receiver()` / `get_subscription_receiver()` |
50 
51## Send Messages (Async)
52 
53```python
54import asyncio
55from azure.servicebus.aio import ServiceBusClient
56from azure.servicebus import ServiceBusMessage
57from azure.identity.aio import DefaultAzureCredential
58 
59async def send_messages():
60    credential = DefaultAzureCredential()
61    
62    async with ServiceBusClient(
63        fully_qualified_namespace="<namespace>.servicebus.windows.net",
64        credential=credential
65    ) as client:
66        sender = client.get_queue_sender(queue_name="myqueue")
67        
68        async with sender:
69            # Single message
70            message = ServiceBusMessage("Hello, Service Bus!")
71            await sender.send_messages(message)
72            
73            # Batch of messages
74            messages = [ServiceBusMessage(f"Message {i}") for i in range(10)]
75            await sender.send_messages(messages)
76            
77            # Message batch (for size control)
78            batch = await sender.create_message_batch()
79            for i in range(100):
80                try:
81                    batch.add_message(ServiceBusMessage(f"Batch message {i}"))
82                except ValueError:  # Batch full
83                    await sender.send_messages(batch)
84                    batch = await sender.create_message_batch()
85                    batch.add_message(ServiceBusMessage(f"Batch message {i}"))
86            await sender.send_messages(batch)
87 
88asyncio.run(send_messages())
89```
90 
91## Receive Messages (Async)
92 
93```python
94async def receive_messages():
95    credential = DefaultAzureCredential()
96    
97    async with ServiceBusClient(
98        fully_qualified_namespace="<namespace>.servicebus.windows.net",
99        credential=credential
100    ) as client:
101        receiver = client.get_queue_receiver(queue_name="myqueue")
102        
103        async with receiver:
104            # Receive batch
105            messages = await receiver.receive_messages(
106                max_message_count=10,
107                max_wait_time=5  # seconds
108            )
109            
110            for msg in messages:
111                print(f"Received: {str(msg)}")
112                await receiver.complete_message(msg)  # Remove from queue
113 
114asyncio.run(receive_messages())
115```
116 
117## Receive Modes
118 
119| Mode | Behavior | Use Case |
120|------|----------|----------|
121| `PEEK_LOCK` (default) | Message locked, must complete/abandon | Reliable processing |
122| `RECEIVE_AND_DELETE` | Removed immediately on receive | At-most-once delivery |
123 
124```python
125from azure.servicebus import ServiceBusReceiveMode
126 
127receiver = client.get_queue_receiver(
128    queue_name="myqueue",
129    receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE
130)
131```
132 
133## Message Settlement
134 
135```python
136async with receiver:
137    messages = await receiver.receive_messages(max_message_count=1)
138    
139    for msg in messages:
140        try:
141            # Process message...
142            await receiver.complete_message(msg)  # Success - remove from queue
143        except ProcessingError:
144            await receiver.abandon_message(msg)  # Retry later
145        except PermanentError:
146            await receiver.dead_letter_message(
147                msg,
148                reason="ProcessingFailed",
149                error_description="Could not process"
150            )
151```
152 
153| Action | Effect |
154|--------|--------|
155| `complete_message()` | Remove from queue (success) |
156| `abandon_message()` | Release lock, retry immediately |
157| `dead_letter_message()` | Move to dead-letter queue |
158| `defer_message()` | Set aside, receive by sequence number |
159 
160## Topics and Subscriptions
161 
162```python
163# Send to topic
164sender = client.get_topic_sender(topic_name="mytopic")
165async with sender:
166    await sender.send_messages(ServiceBusMessage("Topic message"))
167 
168# Receive from subscription
169receiver = client.get_subscription_receiver(
170    topic_name="mytopic",
171    subscription_name="mysubscription"
172)
173async with receiver:
174    messages = await receiver.receive_messages(max_message_count=10)
175```
176 
177## Sessions (FIFO)
178 
179```python
180# Send with session
181message = ServiceBusMessage("Session message")
182message.session_id = "order-123"
183await sender.send_messages(message)
184 
185# Receive from specific session
186receiver = client.get_queue_receiver(
187    queue_name="session-queue",
188    session_id="order-123"
189)
190 
191# Receive from next available session
192from azure.servicebus import NEXT_AVAILABLE_SESSION
193receiver = client.get_queue_receiver(
194    queue_name="session-queue",
195    session_id=NEXT_AVAILABLE_SESSION
196)
197```
198 
199## Scheduled Messages
200 
201```python
202from datetime import datetime, timedelta, timezone
203 
204message = ServiceBusMessage("Scheduled message")
205scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=10)
206 
207# Schedule message
208sequence_number = await sender.schedule_messages(message, scheduled_time)
209 
210# Cancel scheduled message
211await sender.cancel_scheduled_messages(sequence_number)
212```
213 
214## Dead-Letter Queue
215 
216```python
217from azure.servicebus import ServiceBusSubQueue
218 
219# Receive from dead-letter queue
220dlq_receiver = client.get_queue_receiver(
221    queue_name="myqueue",
222    sub_queue=ServiceBusSubQueue.DEAD_LETTER
223)
224 
225async with dlq_receiver:
226    messages = await dlq_receiver.receive_messages(max_message_count=10)
227    for msg in messages:
228        print(f"Dead-lettered: {msg.dead_letter_reason}")
229        await dlq_receiver.complete_message(msg)
230```
231 
232## Sync Client (for simple scripts)
233 
234```python
235from azure.servicebus import ServiceBusClient, ServiceBusMessage
236from azure.identity import DefaultAzureCredential
237 
238with ServiceBusClient(
239    fully_qualified_namespace="<namespace>.servicebus.windows.net",
240    credential=DefaultAzureCredential()
241) as client:
242    with client.get_queue_sender("myqueue") as sender:
243        sender.send_messages(ServiceBusMessage("Sync message"))
244    
245    with client.get_queue_receiver("myqueue") as receiver:
246        for msg in receiver:
247            print(str(msg))
248            receiver.complete_message(msg)
249```
250 
251## Best Practices
252 
2531. **Use async client** for production workloads
2542. **Use context managers** (`async with`) for proper cleanup
2553. **Complete messages** after successful processing
2564. **Use dead-letter queue** for poison messages
2575. **Use sessions** for ordered, FIFO processing
2586. **Use message batches** for high-throughput scenarios
2597. **Set `max_wait_time`** to avoid infinite blocking
260 
261## Reference Files
262 
263| File | Contents |
264|------|----------|
265| [references/patterns.md](references/patterns.md) | Competing consumers, sessions, retry patterns, request-response, transactions |
266| [references/dead-letter.md](references/dead-letter.md) | DLQ handling, poison messages, reprocessing strategies |
267| [scripts/setup_servicebus.py](scripts/setup_servicebus.py) | CLI for queue/topic/subscription management and DLQ monitoring |
268