Service Bus Message Builder

A utility module for building Azure Service Bus messages with rich metadata and properties for a notification system.

Capabilities

Create notification messages with routing properties

The module provides a function to build notification messages with appropriate routing and identification properties.

• Given notification data with type "email", recipient "user@example.com", and content "Test message", creates a message with message_id, content_type "application/json", correlation_id matching the notification ID, to property set to recipient, and subject "notification". @test
• Given notification data with type "sms" and empty content, creates a message with all standard properties set and body containing empty string. @test

Create messages with custom application properties

The module supports adding custom metadata to messages via application properties.

• Given notification data with priority "high" and tags ["alert", "urgent"], creates a message with application_properties containing "priority": "high" and "tags": ["alert", "urgent"]. @test
• Given notification data with no custom properties, creates a message with empty application_properties dictionary. @test

Create messages with temporal properties

The module supports setting time-based properties for message lifecycle management.

• Given a time_to_live value of 3600 seconds, creates a message with time_to_live set to timedelta(seconds=3600). @test
• Given a scheduled_time 60 seconds in the future, creates a message with scheduled_enqueue_time_utc set to the specified datetime. @test

Create messages with session properties

The module supports session-based messaging for ordered processing.

• Given notification data with session_id "user-123" and reply_to_session_id "processor-456", creates a message with both session properties set. @test
• Given notification data with only session_id "user-789", creates a message with session_id set and reply_to_session_id as None. @test

Implementation

@generates

API

from typing import Dict, Any, Optional
from datetime import datetime, timedelta
from azure.servicebus import ServiceBusMessage

def build_notification_message(
    notification_data: Dict[str, Any],
    time_to_live: Optional[int] = None,
    scheduled_time: Optional[datetime] = None
) -> ServiceBusMessage:
    """
    Build a Service Bus message from notification data with appropriate properties.

    Args:
        notification_data: Dictionary containing:
            - type: str - notification type (e.g., "email", "sms")
            - recipient: str - recipient identifier
            - content: str - notification content
            - notification_id: str - unique notification identifier
            - priority: Optional[str] - priority level
            - tags: Optional[List[str]] - tags for categorization
            - session_id: Optional[str] - session identifier for ordered processing
            - reply_to_session_id: Optional[str] - reply-to session identifier
        time_to_live: Optional time-to-live in seconds
        scheduled_time: Optional scheduled delivery time

    Returns:
        ServiceBusMessage with all properties configured
    """
    pass

Dependencies { .dependencies }

azure-servicebus { .dependency }

Provides Azure Service Bus messaging client and message types.

@satisfied-by