or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authorization-security.mddisaster-recovery.mdindex.mdmessage-filtering-rules.mdmigration-monitoring.mdnamespace-management.mdqueue-operations.mdtopic-subscription-management.md
tile.json

tessl/pypi-azure-mgmt-servicebus

Microsoft Azure Service Bus Management Client Library for programmatic control of Service Bus namespaces, queues, topics, subscriptions, and authorization rules

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/azure-mgmt-servicebus@9.0.x

To install, run

npx @tessl/cli install tessl/pypi-azure-mgmt-servicebus@9.0.0

index.mddocs/

Azure Service Bus Management Client

A comprehensive Python client library for managing Azure Service Bus resources through the Azure Resource Manager API. This library enables programmatic creation, configuration, and management of Service Bus namespaces, queues, topics, subscriptions, and authorization rules within Azure subscriptions.

Package Information

  • Package Name: azure-mgmt-servicebus
  • Language: Python
  • Installation: pip install azure-mgmt-servicebus
  • Additional Requirements: pip install azure-identity (for authentication)

Core Imports

from azure.mgmt.servicebus import ServiceBusManagementClient
from azure.identity import DefaultAzureCredential
from azure.core.polling import LROPoller

Async client:

from azure.mgmt.servicebus.aio import ServiceBusManagementClient
from azure.identity.aio import DefaultAzureCredential
from azure.core.polling import LROPoller

Basic Usage

from azure.mgmt.servicebus import ServiceBusManagementClient
from azure.identity import DefaultAzureCredential
import os

# Initialize the client
credential = DefaultAzureCredential()
subscription_id = os.getenv("AZURE_SUBSCRIPTION_ID")
client = ServiceBusManagementClient(credential, subscription_id)

# Create a namespace
namespace_params = {
    "location": "East US",
    "sku": {
        "name": "Standard",
        "tier": "Standard"
    }
}

namespace_operation = client.namespaces.begin_create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    parameters=namespace_params
)

# Wait for the operation to complete
namespace = namespace_operation.result()

# Create a queue
queue_params = {
    "max_size_in_megabytes": 5120,
    "default_message_time_to_live": "P14D"  # 14 days
}

queue = client.queues.create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    queue_name="my-queue",
    parameters=queue_params
)

# Close the client
client.close()

Architecture

The Azure Service Bus Management Client follows the Azure SDK design principles:

  • Multi-API Version Support: Supports multiple Service Bus API versions (2021-11-01 default, 2022-10-01-preview)
  • Operation Groups: Organizes functionality into logical operation groups (namespaces, queues, topics, etc.)
  • ARM Integration: Built on Azure Resource Manager (ARM) patterns with long-running operation support
  • Authentication: Uses Azure Active Directory token-based authentication via azure.identity
  • Async Support: Provides both synchronous and asynchronous client implementations

Capabilities

Namespace Management

Comprehensive namespace lifecycle management including creation, configuration, monitoring, network security, and disaster recovery setup.

class NamespacesOperations:
    def list(self) -> ItemPaged[SBNamespace]: ...
    def list_by_resource_group(self, resource_group_name: str) -> ItemPaged[SBNamespace]: ...
    def begin_create_or_update(self, resource_group_name: str, namespace_name: str, parameters: SBNamespace) -> LROPoller[SBNamespace]: ...
    def get(self, resource_group_name: str, namespace_name: str) -> SBNamespace: ...
    def begin_delete(self, resource_group_name: str, namespace_name: str) -> LROPoller[None]: ...
    def check_name_availability(self, parameters: CheckNameAvailability) -> CheckNameAvailabilityResult: ...

Namespace Management

Queue Operations

Queue lifecycle management, configuration, and authorization rule management for point-to-point messaging scenarios.

class QueuesOperations:
    def list_by_namespace(self, resource_group_name: str, namespace_name: str) -> ItemPaged[SBQueue]: ...
    def create_or_update(self, resource_group_name: str, namespace_name: str, queue_name: str, parameters: SBQueue) -> SBQueue: ...
    def get(self, resource_group_name: str, namespace_name: str, queue_name: str) -> SBQueue: ...
    def delete(self, resource_group_name: str, namespace_name: str, queue_name: str) -> None: ...

Queue Operations

Topic and Subscription Management

Topic creation and management, subscription handling, and message filtering rules for publish-subscribe messaging patterns.

class TopicsOperations:
    def list_by_namespace(self, resource_group_name: str, namespace_name: str) -> ItemPaged[SBTopic]: ...
    def create_or_update(self, resource_group_name: str, namespace_name: str, topic_name: str, parameters: SBTopic) -> SBTopic: ...
    def get(self, resource_group_name: str, namespace_name: str, topic_name: str) -> SBTopic: ...
    def delete(self, resource_group_name: str, namespace_name: str, topic_name: str) -> None: ...

class SubscriptionsOperations:
    def list_by_topic(self, resource_group_name: str, namespace_name: str, topic_name: str) -> ItemPaged[SBSubscription]: ...
    def create_or_update(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str, parameters: SBSubscription) -> SBSubscription: ...
    def get(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str) -> SBSubscription: ...
    def delete(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str) -> None: ...

Topic and Subscription Management

Message Filtering Rules

Message filtering and routing configuration using SQL filters and correlation filters for subscription-based message processing.

class RulesOperations:
    def list_by_subscriptions(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str) -> ItemPaged[Rule]: ...
    def create_or_update(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str, rule_name: str, parameters: Rule) -> Rule: ...
    def get(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str, rule_name: str) -> Rule: ...
    def delete(self, resource_group_name: str, namespace_name: str, topic_name: str, subscription_name: str, rule_name: str) -> None: ...

Message Filtering Rules

Authorization and Security

Access control through authorization rules, access key management, private endpoint configuration, and network security rules.

def list_authorization_rules(self, resource_group_name: str, namespace_name: str) -> ItemPaged[SBAuthorizationRule]: ...
def create_or_update_authorization_rule(self, resource_group_name: str, namespace_name: str, authorization_rule_name: str, parameters: SBAuthorizationRule) -> SBAuthorizationRule: ...
def list_keys(self, resource_group_name: str, namespace_name: str, authorization_rule_name: str) -> AccessKeys: ...
def regenerate_keys(self, resource_group_name: str, namespace_name: str, authorization_rule_name: str, parameters: RegenerateAccessKeyParameters) -> AccessKeys: ...

Authorization and Security

Disaster Recovery

Geo-disaster recovery configuration, pairing management, failover operations, and data replication control.

class DisasterRecoveryConfigsOperations:
    def list(self, resource_group_name: str, namespace_name: str) -> ItemPaged[ArmDisasterRecovery]: ...
    def create_or_update(self, resource_group_name: str, namespace_name: str, alias: str, parameters: ArmDisasterRecovery) -> ArmDisasterRecovery: ...
    def get(self, resource_group_name: str, namespace_name: str, alias: str) -> ArmDisasterRecovery: ...
    def break_pairing(self, resource_group_name: str, namespace_name: str, alias: str) -> None: ...
    def fail_over(self, resource_group_name: str, namespace_name: str, alias: str) -> None: ...

Disaster Recovery

Migration and Monitoring

Standard to Premium namespace migration operations and operational monitoring through operation listings and private networking.

class MigrationConfigsOperations:
    def create_and_start_migration(self, resource_group_name: str, namespace_name: str, config_name: str, parameters: MigrationConfigProperties) -> LROPoller[MigrationConfigProperties]: ...
    def complete_migration(self, resource_group_name: str, namespace_name: str, config_name: str) -> None: ...
    def revert(self, resource_group_name: str, namespace_name: str, config_name: str) -> None: ...

class Operations:
    def list(self) -> ItemPaged[Operation]: ...

Migration and Monitoring

Core Types

class ServiceBusManagementClient:
    def __init__(
        self,
        credential: TokenCredential,
        subscription_id: str,
        api_version: Optional[str] = None,
        base_url: Optional[str] = None,
        **kwargs: Any
    ): ...
    
    # Operation group properties
    disaster_recovery_configs: DisasterRecoveryConfigsOperations
    migration_configs: MigrationConfigsOperations
    namespaces: NamespacesOperations
    operations: Operations
    private_endpoint_connections: PrivateEndpointConnectionsOperations
    private_link_resources: PrivateLinkResourcesOperations
    queues: QueuesOperations
    rules: RulesOperations
    subscriptions: SubscriptionsOperations
    topics: TopicsOperations
    
    def close(self) -> None: ...
    def __enter__(self) -> "ServiceBusManagementClient": ...
    def __exit__(self, *exc_details) -> None: ...

class SBNamespace:
    def __init__(self, **kwargs): ...
    
    # Standard Azure resource properties
    id: Optional[str]
    name: Optional[str] 
    type: Optional[str]
    location: Optional[str]
    tags: Optional[Dict[str, str]]
    
    # Service Bus specific properties
    sku: Optional[SBSku]
    identity: Optional[Identity]
    provisioning_state: Optional[str]
    status: Optional[str]
    created_at: Optional[datetime]
    updated_at: Optional[datetime]
    service_bus_endpoint: Optional[str]
    zone_redundant: Optional[bool]
    encryption: Optional[Encryption]
    private_endpoint_connections: Optional[List[PrivateEndpointConnection]]
    disable_local_auth: Optional[bool]
    alternate_name: Optional[str]

class SBSku:
    def __init__(self, **kwargs): ...
    
    name: Union[str, SkuName]  # "Basic", "Standard", "Premium"
    tier: Optional[Union[str, SkuTier]]  # "Basic", "Standard", "Premium"
    capacity: Optional[int]  # 1, 2, 4 for Premium; ignored for Basic/Standard

from enum import Enum

class SkuName(str, Enum):
    BASIC = "Basic"
    STANDARD = "Standard"
    PREMIUM = "Premium"

class SkuTier(str, Enum):
    BASIC = "Basic"
    STANDARD = "Standard"
    PREMIUM = "Premium"

class EntityStatus(str, Enum):
    ACTIVE = "Active"
    DISABLED = "Disabled"
    RESTORING = "Restoring"
    SEND_DISABLED = "SendDisabled"
    RECEIVE_DISABLED = "ReceiveDisabled"
    CREATING = "Creating"
    DELETING = "Deleting"
    RENAMING = "Renaming"
    UNKNOWN = "Unknown"

class AccessRights(str, Enum):
    MANAGE = "Manage"
    SEND = "Send"
    LISTEN = "Listen"

Error Handling

The client uses standard Azure SDK error handling patterns. Common exceptions include:

  • HttpResponseError: For HTTP-level errors (4xx, 5xx responses)
  • ResourceNotFoundError: When requested resources don't exist
  • ResourceExistsError: When trying to create resources that already exist
  • ClientAuthenticationError: For authentication failures
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError

try:
    namespace = client.namespaces.get("resource-group", "namespace-name")
except ResourceNotFoundError:
    print("Namespace not found")
except HttpResponseError as e:
    print(f"HTTP error: {e.status_code} - {e.message}")