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

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Namespace Management

Name: tessl/pypi-azure-mgmt-servicebus
Author: tessl

Comprehensive namespace lifecycle management including creation, configuration, monitoring, network security, and disaster recovery setup. Namespaces are the top-level containers for all Service Bus messaging entities and provide the security boundary for access control.

Capabilities

Namespace Lifecycle Operations

Create, retrieve, update, delete, and list Service Bus namespaces across subscriptions and resource groups.

def list(self) -> ItemPaged[SBNamespace]:
    """List all namespaces in the subscription."""

def list_by_resource_group(self, resource_group_name: str) -> ItemPaged[SBNamespace]:
    """List namespaces in a specific resource group.
    
    Args:
        resource_group_name (str): Name of the resource group.
    
    Returns:
        ItemPaged[SBNamespace]: Iterable of namespace resources.
    """

def begin_create_or_update(
    self,
    resource_group_name: str,
    namespace_name: str,
    parameters: SBNamespace
) -> LROPoller[SBNamespace]:
    """Create or update a Service Bus namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        parameters (SBNamespace): Namespace configuration parameters.
    
    Returns:
        LROPoller[SBNamespace]: A poller for the long-running operation.
    """

def get(self, resource_group_name: str, namespace_name: str) -> SBNamespace:
    """Get details of a specific namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        SBNamespace: The namespace resource.
    """

def begin_delete(
    self,
    resource_group_name: str,
    namespace_name: str
) -> LROPoller[None]:
    """Delete a namespace (long-running operation).
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        LROPoller[None]: Long-running operation poller.
    """

def update(
    self,
    resource_group_name: str,
    namespace_name: str,
    parameters: SBNamespaceUpdateParameters
) -> SBNamespace:
    """Update namespace properties.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        parameters (SBNamespaceUpdateParameters): Update parameters.
    
    Returns:
        SBNamespace: The updated namespace.
    """

Name Availability Checking

Verify namespace name availability before creation to prevent conflicts.

def check_name_availability(
    self,
    parameters: CheckNameAvailability
) -> CheckNameAvailabilityResult:
    """Check if a namespace name is available.
    
    Args:
        parameters (CheckNameAvailability): Name availability check request.
    
    Returns:
        CheckNameAvailabilityResult: Availability result with reason if unavailable.
    """

Authorization Rule Management

Manage shared access signature (SAS) authorization rules for namespace-level access control.

def list_authorization_rules(
    self,
    resource_group_name: str,
    namespace_name: str
) -> ItemPaged[SBAuthorizationRule]:
    """List authorization rules for the namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        ItemPaged[SBAuthorizationRule]: Iterable of authorization rules.
    """

def create_or_update_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str,
    parameters: SBAuthorizationRule
) -> SBAuthorizationRule:
    """Create or update an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
        parameters (SBAuthorizationRule): Authorization rule parameters.
    
    Returns:
        SBAuthorizationRule: The created or updated authorization rule.
    """

def get_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> SBAuthorizationRule:
    """Get an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    
    Returns:
        SBAuthorizationRule: The authorization rule.
    """

def delete_authorization_rule(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> None:
    """Delete an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    """

Access Key Management

Generate and manage access keys for namespace authentication.

def list_keys(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str
) -> AccessKeys:
    """Get access keys for an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
    
    Returns:
        AccessKeys: Primary and secondary keys with connection strings.
    """

def regenerate_keys(
    self,
    resource_group_name: str,
    namespace_name: str,
    authorization_rule_name: str,
    parameters: RegenerateAccessKeyParameters
) -> AccessKeys:
    """Regenerate access keys for an authorization rule.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        authorization_rule_name (str): Name of the authorization rule.
        parameters (RegenerateAccessKeyParameters): Regeneration parameters.
    
    Returns:
        AccessKeys: New keys with connection strings.
    """

Network Security Configuration

Configure network access rules, IP filtering, and virtual network integration.

def create_or_update_network_rule_set(
    self,
    resource_group_name: str,
    namespace_name: str,
    parameters: NetworkRuleSet
) -> NetworkRuleSet:
    """Create or update network rule set for the namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
        parameters (NetworkRuleSet): Network rule configuration.
    
    Returns:
        NetworkRuleSet: The network rule set.
    """

def get_network_rule_set(
    self,
    resource_group_name: str,
    namespace_name: str
) -> NetworkRuleSet:
    """Get network rule set for the namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        NetworkRuleSet: The current network rule set.
    """

def list_network_rule_sets(
    self,
    resource_group_name: str,
    namespace_name: str
) -> ItemPaged[NetworkRuleSet]:
    """Gets list of NetworkRuleSet for a Namespace.
    
    Args:
        resource_group_name (str): Name of the resource group.
        namespace_name (str): Name of the namespace.
    
    Returns:
        ItemPaged[NetworkRuleSet]: An iterator of network rule sets.
    """

Usage Examples

Creating a Basic Namespace

from azure.mgmt.servicebus import ServiceBusManagementClient
from azure.mgmt.servicebus.models import SBNamespace, SBSku
from azure.identity import DefaultAzureCredential

client = ServiceBusManagementClient(DefaultAzureCredential(), subscription_id)

# Create a Standard tier namespace
namespace_params = SBNamespace(
    location="East US",
    sku=SBSku(
        name="Standard",
        tier="Standard"
    ),
    tags={"environment": "production", "team": "messaging"}
)

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()

print(f"Namespace created: {namespace.name}")
print(f"Service Bus endpoint: {namespace.service_bus_endpoint}")

Creating a Premium Namespace with Zone Redundancy

# Create a Premium tier namespace with zone redundancy
premium_params = SBNamespace(
    location="East US",
    sku=SBSku(
        name="Premium",
        tier="Premium",
        capacity=1  # 1, 2, or 4 messaging units
    ),
    zone_redundant=True,
    tags={"environment": "production", "tier": "premium"}
)

premium_namespace_operation = client.namespaces.begin_create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-premium-namespace",
    parameters=premium_params
)

# Wait for the operation to complete
premium_namespace = premium_namespace_operation.result()

Creating Authorization Rules

from azure.mgmt.servicebus.models import SBAuthorizationRule, AccessRights

# Create a send-only authorization rule
send_rule = SBAuthorizationRule(
    rights=[AccessRights.SEND]
)

auth_rule = client.namespaces.create_or_update_authorization_rule(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="SendOnlyPolicy",
    parameters=send_rule
)

# Get the access keys
keys = client.namespaces.list_keys(
    resource_group_name="my-resource-group",
    namespace_name="my-servicebus-namespace",
    authorization_rule_name="SendOnlyPolicy"
)

print(f"Primary connection string: {keys.primary_connection_string}")

Checking Name Availability

from azure.mgmt.servicebus.models import CheckNameAvailability

# Check if a namespace name is available
check_params = CheckNameAvailability(name="my-unique-namespace")
result = client.namespaces.check_name_availability(check_params)

if result.name_available:
    print("Namespace name is available")
else:
    print(f"Name unavailable: {result.reason} - {result.message}")

Types

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 SBNamespaceUpdateParameters:
    def __init__(self, **kwargs): ...
    
    location: Optional[str]
    tags: Optional[Dict[str, str]]
    sku: Optional[SBSku]
    identity: Optional[Identity]
    encryption: Optional[Encryption]
    disable_local_auth: Optional[bool]
    alternate_name: Optional[str]

class CheckNameAvailability:
    def __init__(self, **kwargs): ...
    
    name: str

class CheckNameAvailabilityResult:
    def __init__(self, **kwargs): ...
    
    message: Optional[str]
    name_available: Optional[bool]
    reason: Optional[Union[str, UnavailableReason]]

class UnavailableReason(str, Enum):
    NONE = "None"
    INVALID_NAME = "InvalidName"
    SUBSCRIPTION_IS_DISABLED = "SubscriptionIsDisabled"
    NAME_IN_USE = "NameInUse"
    NAME_IN_LOCKDOWN = "NameInLockdown"
    TOO_MANY_NAMESPACE_IN_CURRENT_SUBSCRIPTION = "TooManyNamespaceInCurrentSubscription"

class NetworkRuleSet:
    def __init__(self, **kwargs): ...
    
    trusted_service_access_enabled: Optional[bool]
    default_action: Optional[Union[str, DefaultAction]]
    virtual_network_rules: Optional[List[NWRuleSetVirtualNetworkRules]]
    ip_rules: Optional[List[NWRuleSetIpRules]]

class RegenerateAccessKeyParameters:
    def __init__(self, **kwargs): ...
    
    key_type: Union[str, KeyType]  # "PrimaryKey" or "SecondaryKey"
    key: Optional[str]

class KeyType(str, Enum):
    PRIMARY_KEY = "PrimaryKey"
    SECONDARY_KEY = "SecondaryKey"

Install with Tessl CLI

npx tessl i tessl/pypi-azure-mgmt-servicebus

docs

authorization-security.md

disaster-recovery.md

index.md

message-filtering-rules.md

migration-monitoring.md

namespace-management.md

queue-operations.md

topic-subscription-management.md

tile.json

tessl/pypi-azure-mgmt-servicebus