tessl/pypi-azure-mgmt-iothub

Microsoft Azure IoT Hub Management Client Library for programmatic management of Azure IoT Hub resources through the Azure Resource Manager API

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Private Networking

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

Private endpoint and private link connectivity management for secure IoT Hub access through virtual networks, enabling isolated network communication and enhanced security for enterprise IoT deployments with controlled access patterns and network segmentation.

Capabilities

Private Endpoint Connections

Manage private endpoint connections that enable secure, private connectivity to IoT Hub through Azure Virtual Networks without exposing traffic to the public internet.

def list(resource_group_name: str, resource_name: str, **kwargs) -> List[PrivateEndpointConnection]:
    """
    List all private endpoint connections for the IoT hub.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        
    Returns:
        List[PrivateEndpointConnection]: All private endpoint connections with status and configuration
    """

def get(
    resource_group_name: str, 
    resource_name: str, 
    private_endpoint_connection_name: str, 
    **kwargs
) -> PrivateEndpointConnection:
    """
    Get details of a specific private endpoint connection.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        private_endpoint_connection_name: Name of the private endpoint connection
        
    Returns:
        PrivateEndpointConnection: Connection details including status and private endpoint information
    """

def begin_update(
    resource_group_name: str, 
    resource_name: str, 
    private_endpoint_connection_name: str, 
    private_endpoint_connection: PrivateEndpointConnection, 
    **kwargs
) -> LROPoller[PrivateEndpointConnection]:
    """
    Update private endpoint connection configuration and approval status.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        private_endpoint_connection_name: Name of the private endpoint connection
        private_endpoint_connection: Updated connection configuration
        
    Returns:
        LROPoller[PrivateEndpointConnection]: Long-running operation for connection update
    """

def begin_delete(
    resource_group_name: str, 
    resource_name: str, 
    private_endpoint_connection_name: str, 
    **kwargs
) -> LROPoller[PrivateEndpointConnection]:
    """
    Delete a private endpoint connection from the IoT hub.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        private_endpoint_connection_name: Name of the private endpoint connection to delete
        
    Returns:
        LROPoller[PrivateEndpointConnection]: Long-running operation for connection deletion
    """

Private Link Resources

Discover and manage private link resources that define the available connection targets and group IDs for private endpoint configuration.

def list(resource_group_name: str, resource_name: str, **kwargs) -> PrivateLinkResources:
    """
    List all available private link resources for the IoT hub.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        
    Returns:
        PrivateLinkResources: Available private link resources and supported group IDs
    """

def get(
    resource_group_name: str, 
    resource_name: str, 
    group_id: str, 
    **kwargs
) -> GroupIdInformation:
    """
    Get information about a specific private link resource group.
    
    Args:
        resource_group_name: Name of the resource group
        resource_name: Name of the IoT hub resource
        group_id: Private link resource group identifier
        
    Returns:
        GroupIdInformation: Details about the private link resource group and requirements
    """

Usage Examples

Setting up private endpoint connectivity

from azure.identity import DefaultAzureCredential
from azure.mgmt.iothub import IotHubClient
from azure.mgmt.iothub.models import (
    PrivateEndpointConnection, 
    PrivateEndpointConnectionProperties,
    PrivateLinkServiceConnectionState
)

# Initialize client
credential = DefaultAzureCredential()
client = IotHubClient(credential, "subscription-id")

resource_group = "myResourceGroup"
hub_name = "myIoTHub"

# Discover available private link resources
print("Available private link resources:")
private_link_resources = client.private_link_resources.list(resource_group, hub_name)
for resource in private_link_resources.value:
    print(f"  - Group ID: {resource.group_id}")
    print(f"    Required members: {resource.required_members}")
    print(f"    Required zone names: {resource.required_zone_names}")

# Get details for IoT Hub private link resource
iot_hub_resource = client.private_link_resources.get(
    resource_group, hub_name, "iotHub"
)
print(f"IoT Hub private link resource: {iot_hub_resource.group_id}")
print(f"Required members: {iot_hub_resource.required_members}")

Managing private endpoint connections

# List all private endpoint connections
print("Current private endpoint connections:")
connections = client.private_endpoint_connections.list(resource_group, hub_name)
for connection in connections:
    print(f"  - Name: {connection.name}")
    print(f"    Status: {connection.properties.private_link_service_connection_state.status}")
    print(f"    Description: {connection.properties.private_link_service_connection_state.description}")
    if connection.properties.private_endpoint:
        print(f"    Private endpoint ID: {connection.properties.private_endpoint.id}")

# Approve a pending private endpoint connection
pending_connection_name = "hub-private-endpoint_12345"
try:
    connection = client.private_endpoint_connections.get(
        resource_group, hub_name, pending_connection_name
    )
    
    if connection.properties.private_link_service_connection_state.status == "Pending":
        # Approve the connection
        connection.properties.private_link_service_connection_state.status = "Approved"
        connection.properties.private_link_service_connection_state.description = "Approved for production use"
        
        update_operation = client.private_endpoint_connections.begin_update(
            resource_group, hub_name, pending_connection_name, connection
        )
        updated_connection = update_operation.result()
        print(f"Approved connection: {updated_connection.name}")
    else:
        print(f"Connection is already {connection.properties.private_link_service_connection_state.status}")
        
except Exception as e:
    print(f"Connection not found or error occurred: {e}")

Rejecting and cleaning up private endpoints

# Reject a private endpoint connection
def reject_private_endpoint(connection_name: str, reason: str):
    try:
        connection = client.private_endpoint_connections.get(
            resource_group, hub_name, connection_name
        )
        
        # Reject the connection
        connection.properties.private_link_service_connection_state.status = "Rejected"
        connection.properties.private_link_service_connection_state.description = reason
        
        update_operation = client.private_endpoint_connections.begin_update(
            resource_group, hub_name, connection_name, connection
        )
        updated_connection = update_operation.result()
        print(f"Rejected connection: {updated_connection.name}")
        print(f"Reason: {reason}")
        
    except Exception as e:
        print(f"Failed to reject connection {connection_name}: {e}")

# Reject unauthorized connection attempts
reject_private_endpoint(
    "unauthorized-endpoint_67890", 
    "Connection not authorized for this environment"
)

# Delete old or unused private endpoint connections
connections_to_delete = ["old-dev-endpoint_11111", "test-endpoint_22222"]

for connection_name in connections_to_delete:
    try:
        delete_operation = client.private_endpoint_connections.begin_delete(
            resource_group, hub_name, connection_name
        )
        delete_operation.result()  # Wait for completion
        print(f"Deleted private endpoint connection: {connection_name}")
        
    except Exception as e:
        print(f"Failed to delete connection {connection_name}: {e}")

Monitoring private endpoint status

def monitor_private_endpoints(resource_group: str, hub_name: str):
    """Monitor and report on all private endpoint connections."""
    
    connections = client.private_endpoint_connections.list(resource_group, hub_name)
    
    status_counts = {"Approved": 0, "Pending": 0, "Rejected": 0, "Disconnected": 0}
    connection_details = []
    
    for connection in connections:
        status = connection.properties.private_link_service_connection_state.status
        status_counts[status] = status_counts.get(status, 0) + 1
        
        connection_info = {
            "name": connection.name,
            "status": status,
            "description": connection.properties.private_link_service_connection_state.description,
            "endpoint_id": connection.properties.private_endpoint.id if connection.properties.private_endpoint else None
        }
        connection_details.append(connection_info)
    
    # Print summary
    print(f"Private Endpoint Connection Summary for {hub_name}:")
    print(f"  Total connections: {len(connection_details)}")
    for status, count in status_counts.items():
        if count > 0:
            print(f"  {status}: {count}")
    
    # Print details
    print("\nConnection Details:")
    for conn in connection_details:
        print(f"  - {conn['name']}")
        print(f"    Status: {conn['status']}")
        print(f"    Description: {conn['description']}")
        if conn['endpoint_id']:
            print(f"    Endpoint: {conn['endpoint_id']}")
    
    return connection_details

# Monitor connections
monitor_private_endpoints(resource_group, hub_name)

Types

PrivateEndpointConnection

Represents a private endpoint connection with status, configuration, and metadata for network isolation management.

class PrivateEndpointConnection:
    """Private endpoint connection details."""
    name: str                                           # Connection name
    type: str                                           # Resource type
    properties: PrivateEndpointConnectionProperties     # Connection properties and status

PrivateEndpointConnectionProperties

Configuration and status properties for private endpoint connections including approval state and network details.

class PrivateEndpointConnectionProperties:
    """Private endpoint connection configuration."""
    private_endpoint: PrivateEndpoint                           # Private endpoint resource reference
    private_link_service_connection_state: PrivateLinkServiceConnectionState  # Connection approval status
    provisioning_state: str                                     # Provisioning status

PrivateLinkServiceConnectionState

Connection approval status and description for managing private endpoint access control and lifecycle.

class PrivateLinkServiceConnectionState:
    """Private link service connection state."""
    status: str         # Connection status: Pending, Approved, Rejected, Disconnected
    description: str    # Status description and approval reason
    actions_required: str  # Required actions for connection setup

PrivateLinkResources

Collection of available private link resources and supported group identifiers for connection configuration.

class PrivateLinkResources:
    """Available private link resources."""
    value: List[GroupIdInformation]  # List of available private link resource groups

GroupIdInformation

Information about specific private link resource groups including required members and DNS zone configurations.

class GroupIdInformation:
    """Private link resource group information."""
    group_id: str                    # Private link resource group identifier
    required_members: List[str]      # Required member names for the group
    required_zone_names: List[str]   # Required private DNS zone names

Install with Tessl CLI

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

docs

device-operations.md

event-hub-consumer-groups.md

failover-operations.md

index.md

message-routing.md

monitoring-quotas.md

private-networking.md

resource-management.md

security-management.md

utility-operations.md

tile.json

tessl/pypi-azure-mgmt-iothub