or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

core-management.mdenterprise-features.mdindex.mdsecurity-access-control.md
tile.json

tessl/pypi-azure-mgmt-eventhub

Microsoft Azure Event Hub Management Client Library for Python

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

To install, run

npx @tessl/cli install tessl/pypi-azure-mgmt-eventhub@11.2.0

index.mddocs/

Azure Event Hub Management Client

A comprehensive Python management client library for Azure Event Hub services, enabling developers to programmatically manage Event Hub namespaces, event hubs, consumer groups, authorization rules, disaster recovery configurations, and dedicated clusters through the Azure Resource Manager API.

Package Information

  • Package Name: azure-mgmt-eventhub
  • Language: Python
  • Installation: pip install azure-mgmt-eventhub
  • Version: 11.2.0

Core Imports

from azure.mgmt.eventhub import EventHubManagementClient

Basic Usage

from azure.mgmt.eventhub import EventHubManagementClient
from azure.identity import DefaultAzureCredential

# Initialize the management client
credential = DefaultAzureCredential()
subscription_id = "your-subscription-id"
client = EventHubManagementClient(credential, subscription_id)

# Create a namespace
from azure.mgmt.eventhub.models import EHNamespace, Sku

namespace_params = EHNamespace(
    location="East US",
    sku=Sku(name="Standard", tier="Standard")
)

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

# Create an Event Hub
from azure.mgmt.eventhub.models import Eventhub

eventhub_params = Eventhub(
    message_retention_in_days=7,
    partition_count=4
)

eventhub = client.event_hubs.create_or_update(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace",
    event_hub_name="my-eventhub",
    parameters=eventhub_params
)

# List Event Hubs in namespace
eventhubs = client.event_hubs.list_by_namespace(
    resource_group_name="my-resource-group",
    namespace_name="my-eventhub-namespace"
)

for eh in eventhubs:
    print(f"Event Hub: {eh.name}")

Architecture

The client uses Azure SDK's multi-API version architecture, supporting multiple API versions with 2024-01-01 as the default. Key components include:

  • EventHubManagementClient: Main client class providing access to all operation groups
  • Operation Groups: Specialized managers for different resource types (namespaces, event_hubs, clusters, etc.)
  • Models: Rich data models representing Azure Event Hub resources and configuration
  • Multi-API Support: Backward compatibility with older API versions (2015-08-01 through 2024-05-01-preview)
  • Async Support: Full async/await support through aio submodules

Note: Not all operation groups are available in every API version. The default API version 2024-01-01 includes all currently documented operations. For legacy operations like regions, you may need to specify an older API version like 2017-04-01.

Capabilities

Core Management

Primary Event Hub resource management including namespaces, Event Hubs, and consumer groups. These are the fundamental building blocks for Event Hub solutions.

class EventHubManagementClient:
    def __init__(
        self,
        credential: TokenCredential,
        subscription_id: str,
        api_version: Optional[str] = None,
        base_url: str = "https://management.azure.com",
        profile: KnownProfiles = KnownProfiles.default,
        **kwargs: Any
    ): ...
    
    @property
    def namespaces(self) -> NamespacesOperations: ...
    
    @property
    def event_hubs(self) -> EventHubsOperations: ...
    
    @property
    def consumer_groups(self) -> ConsumerGroupsOperations: ...

Core Management

Security and Access Control

Comprehensive security management including authorization rules, private endpoints, network rules, and managed identity configuration for secure Event Hub access.

class AuthorizationRule:
    def __init__(
        self,
        rights: Optional[List[Union[str, AccessRights]]] = None,
        **kwargs: Any
    ): ...

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

Security and Access Control

Enterprise Features

Advanced enterprise capabilities including dedicated clusters, disaster recovery, schema registry, and application groups for production-scale Event Hub deployments.

@property
def clusters(self) -> ClustersOperations: ...

@property
def disaster_recovery_configs(self) -> DisasterRecoveryConfigsOperations: ...

@property
def schema_registry(self) -> SchemaRegistryOperations: ...

@property
def application_group(self) -> ApplicationGroupOperations: ...

Enterprise Features

Network Security Perimeter

Advanced network security perimeter management for controlling access boundaries and resource associations in enterprise environments.

@property
def network_security_perimeter_configuration(self) -> NetworkSecurityPerimeterConfigurationOperations:
    """Network security perimeter configuration operations."""

@property
def network_security_perimeter_configurations(self) -> NetworkSecurityPerimeterConfigurationsOperations:
    """Network security perimeter configurations operations."""

Operations:

  • network_security_perimeter_configuration.list(resource_group_name, namespace_name, **kwargs) -> Iterable[NetworkSecurityPerimeterConfiguration] - List network security perimeter configurations
  • network_security_perimeter_configurations.begin_create_or_update(resource_group_name, namespace_name, resource_association_name, **kwargs) -> LROPoller[None] - Create or update network security perimeter configuration

Utility Operations

Service discovery operations for listing available Azure Event Hub operations.

@property
def operations(self) -> Operations:
    """List available Event Hub management operations."""

Operations:

  • operations.list(**kwargs) -> Iterable[Operation] - List all available REST API operations

Error Handling

The client raises standard Azure SDK exceptions:

from azure.core.exceptions import (
    HttpResponseError,
    ResourceNotFoundError,
    ResourceExistsError,
    ClientAuthenticationError
)

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

Types

from typing import Optional, List, Union, Any
from azure.core.credentials import TokenCredential
from azure.profiles import KnownProfiles

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

class SkuName(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"

# Core model base classes
class Resource:
    def __init__(self, **kwargs: Any): ...
    
    id: Optional[str]
    name: Optional[str]
    type: Optional[str]

class TrackedResource(Resource):
    def __init__(
        self,
        location: Optional[str] = None,
        tags: Optional[Dict[str, str]] = None,
        **kwargs: Any
    ): ...
    
    location: Optional[str]
    tags: Optional[Dict[str, str]]

class NetworkSecurityPerimeterConfiguration:
    def __init__(
        self,
        network_security_perimeter: Optional[NetworkSecurityPerimeter] = None,
        resource_association: Optional[ResourceAssociation] = None,
        profile: Optional[NetworkSecurityPerimeterConfigurationProfile] = None,
        **kwargs: Any
    ): ...