Azure Service Bus

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient

credential = DefaultAzureCredential()
client = ServiceBusClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=credential,
)

# BAD: connection string embeds shared access key
client = ServiceBusClient.from_connection_string(
    "Endpoint=sb://myns.servicebus.windows.net/;SharedAccessKeyName=..."
)

# GOOD: keyless via RBAC role (Azure Service Bus Data Sender / Receiver)
client = ServiceBusClient(
    fully_qualified_namespace=os.environ["SERVICEBUS_NAMESPACE"],
    credential=DefaultAzureCredential(),
)

Queue:      Producer → [Queue] → Consumer A
Topic:      Producer → [Topic] → [Sub: orders-billing]    → Billing Service
                               → [Sub: orders-shipping]   → Shipping Service
                               → [Sub: orders-analytics]  → Analytics Service

from azure.servicebus import ServiceBusMessage
import json

def send_message(namespace: str, queue_name: str, body: dict, **props) -> None:
    credential = DefaultAzureCredential()
    with ServiceBusClient(namespace, credential) as client:
        with client.get_queue_sender(queue_name) as sender:
            message = ServiceBusMessage(
                json.dumps(body),
                subject=props.get("subject"),
                correlation_id=props.get("correlation_id"),
                message_id=props.get("message_id"),        # deduplication key
                time_to_live=props.get("ttl"),             # timedelta
                application_properties=props.get("properties", {}),
            )
            sender.send_messages(message)

def send_batch(namespace: str, queue_name: str, messages: list[dict]) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            batch = sender.create_message_batch()
            for body in messages:
                try:
                    batch.add_message(ServiceBusMessage(json.dumps(body)))
                except ValueError:
                    # Batch full — send current batch and start a new one
                    sender.send_messages(batch)
                    batch = sender.create_message_batch()
                    batch.add_message(ServiceBusMessage(json.dumps(body)))
            if len(batch):
                sender.send_messages(batch)

from datetime import datetime, timedelta, timezone

def schedule_message(namespace: str, queue_name: str, body: dict, delay: timedelta) -> int:
    enqueue_at = datetime.now(timezone.utc) + delay
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            seq_numbers = sender.schedule_messages(
                ServiceBusMessage(json.dumps(body)),
                enqueue_at,
            )
            return seq_numbers[0]  # use to cancel with cancel_scheduled_messages()

def publish_event(namespace: str, topic_name: str, event_type: str, payload: dict) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_topic_sender(topic_name) as sender:
            sender.send_messages(ServiceBusMessage(
                json.dumps(payload),
                subject=event_type,
                application_properties={"event_type": event_type},
            ))

from azure.servicebus import ServiceBusReceiveMode

def process_queue(namespace: str, queue_name: str, handler, max_messages: int = 10) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
            max_wait_time=5,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=max_messages):
                try:
                    body = json.loads(str(msg))
                    handler(body)
                    receiver.complete_message(msg)      # ack — removes from queue
                except Exception as e:
                    if msg.delivery_count >= 3:
                        receiver.dead_letter_message(   # send to DLQ
                            msg,
                            reason="MaxRetriesExceeded",
                            error_description=str(e),
                        )
                    else:
                        receiver.abandon_message(msg)   # nack — requeues with backoff

import threading

def start_consumer(namespace: str, queue_name: str, handler, stop_event: threading.Event) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(queue_name, max_wait_time=5) as receiver:
            while not stop_event.is_set():
                messages = receiver.receive_messages(max_message_count=10, max_wait_time=5)
                if not messages:
                    continue
                for msg in messages:
                    try:
                        handler(json.loads(str(msg)))
                        receiver.complete_message(msg)
                    except Exception as e:
                        logger.exception("message_processing_failed", extra={"delivery_count": msg.delivery_count})
                        receiver.abandon_message(msg)

def process_dlq(namespace: str, queue_name: str) -> list[dict]:
    dlq_path = f"{queue_name}/$deadletterqueue"
    dead_letters = []

    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            dlq_path,
            receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=50):
                dead_letters.append({
                    "body": json.loads(str(msg)),
                    "reason": msg.dead_letter_reason,
                    "description": msg.dead_letter_error_description,
                    "delivery_count": msg.delivery_count,
                    "enqueued_at": msg.enqueued_time_utc,
                })
                receiver.complete_message(msg)  # drain DLQ after inspection

    return dead_letters

# Send with session ID (all messages for the same order stay ordered)
def send_with_session(namespace: str, queue_name: str, session_id: str, body: dict) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            sender.send_messages(ServiceBusMessage(
                json.dumps(body),
                session_id=session_id,
            ))

# Receive a specific session
def process_session(namespace: str, queue_name: str, session_id: str, handler) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            session_id=session_id,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=100):
                handler(json.loads(str(msg)))
                receiver.complete_message(msg)

# Accept the next available session (let Service Bus assign)
def process_next_session(namespace: str, queue_name: str, handler) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            session_id=NEXT_AVAILABLE_SESSION,  # from azure.servicebus
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=100, max_wait_time=10):
                handler(json.loads(str(msg)))
                receiver.complete_message(msg)

from azure.servicebus.management import ServiceBusAdministrationClient, SqlRuleFilter, CorrelationRuleFilter

admin = ServiceBusAdministrationClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=DefaultAzureCredential(),
)

# SQL filter — route by message property
admin.create_rule(
    topic_name="orders",
    subscription_name="orders-eu",
    rule_name="eu-only",
    filter=SqlRuleFilter("Region = 'EU'"),
)

# Correlation filter — cheaper, matches on built-in or application properties
admin.create_rule(
    topic_name="orders",
    subscription_name="orders-priority",
    rule_name="priority-orders",
    filter=CorrelationRuleFilter(
        subject="order.placed",
        application_properties={"priority": "high"},
    ),
)

# Remove the default catch-all rule when adding selective filters
admin.delete_rule("orders", "orders-eu", "$Default")

from azure.servicebus import ServiceBusClient
from azure.core.pipeline.policies import RetryPolicy

# SDK-level retry (network/transient errors)
client = ServiceBusClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=DefaultAzureCredential(),
    retry_total=5,
    retry_backoff_factor=1.5,
    retry_backoff_max=30,
)

# Message-level retry: set max_delivery_count on the queue/subscription
# (default 10 — messages exceeding this are auto-dead-lettered by Service Bus)
admin.update_queue(
    admin.get_queue("my-queue"),
    max_delivery_count=5,
    lock_duration=timedelta(minutes=2),   # must process within this window
)

from azure.servicebus.exceptions import (
    ServiceBusError,
    ServiceBusConnectionError,
    ServiceBusAuthorizationError,
    MessageLockLostError,
    SessionLockLostError,
    MessageAlreadySettled,
)

def safe_receive(receiver, handler, msg) -> None:
    try:
        handler(json.loads(str(msg)))
        receiver.complete_message(msg)
    except MessageLockLostError:
        # Lock expired before we could settle — message will reappear
        logger.warning("lock_expired", message_id=msg.message_id)
    except MessageAlreadySettled:
        # Duplicate settle call — safe to ignore
        pass
    except ServiceBusAuthorizationError:
        # RBAC role missing — fail fast, don't retry
        raise
    except ServiceBusConnectionError as e:
        logger.error("connection_lost", error=str(e))
        raise  # outer loop should reconnect
    except Exception as e:
        logger.exception("handler_failed")
        try:
            receiver.abandon_message(msg)
        except MessageAlreadySettled:
            pass