CtrlK
BlogDocsLog inGet started
Tessl Logo

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

utility-operations.mddocs/

Utility Operations

Essential utility functions for IoT Hub management including resource name availability checking and API operation discovery, providing foundational capabilities for hub deployment planning and API exploration.

Capabilities

Name Availability Checking

Validate IoT Hub name availability before resource creation to prevent deployment failures and ensure compliance with Azure naming requirements and global uniqueness constraints.

def check_name_availability(operation_inputs: OperationInputs, **kwargs) -> IotHubNameAvailabilityInfo:
    """
    Check if an IoT Hub name is available for use in Azure.
    
    Args:
        operation_inputs: Name validation request containing the proposed hub name
        
    Returns:
        IotHubNameAvailabilityInfo: Availability status and detailed reason if unavailable
    """

API Operation Discovery

Discover all available REST API operations for the IoT Hub service, useful for understanding capabilities, building custom tooling, and API exploration.

def list() -> ItemPaged[Operation]:
    """
    List all available IoT Hub management REST API operations.
    
    Returns:
        ItemPaged[Operation]: Complete list of API operations with metadata and descriptions
    """

Usage Examples

Validating IoT Hub names before deployment

from azure.identity import DefaultAzureCredential
from azure.mgmt.iothub import IotHubClient
from azure.mgmt.iothub.models import OperationInputs

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

def validate_hub_name(proposed_name: str) -> tuple[bool, str]:
    """
    Validate if an IoT Hub name is available and meets requirements.
    
    Returns:
        Tuple of (is_available, message)
    """
    
    print(f"Checking availability for IoT Hub name: '{proposed_name}'")
    
    try:
        # Create validation request
        name_request = OperationInputs(name=proposed_name)
        
        # Check name availability
        availability_result = client.iot_hub_resource.check_name_availability(name_request)
        
        print(f"  Name Available: {availability_result.name_available}")
        
        if availability_result.name_available:
            print(f"  ✓ '{proposed_name}' is available for use")
            return True, "Name is available"
        else:
            print(f"  ✗ '{proposed_name}' is not available")
            print(f"  Reason: {availability_result.reason}")
            print(f"  Message: {availability_result.message}")
            return False, f"{availability_result.reason}: {availability_result.message}"
            
    except Exception as e:
        error_msg = f"Failed to check name availability: {e}"
        print(f"  ✗ {error_msg}")
        return False, error_msg

# Test various hub names
test_names = [
    "my-iot-hub-prod",
    "contoso-manufacturing-hub",
    "iot-hub-dev-001", 
    "azure-iot-hub",  # Likely taken
    "my-unique-hub-12345",
    "test"  # Too short
]

print("IoT Hub Name Validation Results:")
print("=" * 45)

available_names = []
unavailable_names = []

for name in test_names:
    is_available, message = validate_hub_name(name)
    
    if is_available:
        available_names.append(name)
    else:
        unavailable_names.append((name, message))
    
    print()  # Add spacing between checks

# Summary
print("Summary:")
print(f"  Available names: {len(available_names)}")
print(f"  Unavailable names: {len(unavailable_names)}")

if available_names:
    print("\nAvailable for use:")
    for name in available_names:
        print(f"  ✓ {name}")

if unavailable_names:
    print("\nNot available:")
    for name, reason in unavailable_names:
        print(f"  ✗ {name} - {reason}")

Generating unique IoT Hub names

import random
import string
from datetime import datetime

def generate_unique_hub_name(base_name: str, max_attempts: int = 10) -> str:
    """
    Generate a unique IoT Hub name by appending random suffixes.
    
    Args:
        base_name: Base name for the IoT Hub
        max_attempts: Maximum number of generation attempts
        
    Returns:
        Available IoT Hub name, or raises exception if none found
    """
    
    print(f"Generating unique IoT Hub name based on: '{base_name}'")
    
    for attempt in range(max_attempts):
        # Generate suffix with random characters
        suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
        candidate_name = f"{base_name}-{suffix}"
        
        # Ensure name meets length requirements (3-50 characters)
        if len(candidate_name) > 50:
            candidate_name = candidate_name[:50]
        elif len(candidate_name) < 3:
            candidate_name = candidate_name + "001"
        
        print(f"  Attempt {attempt + 1}: '{candidate_name}'")
        
        is_available, message = validate_hub_name(candidate_name)
        
        if is_available:
            print(f"  ✓ Generated unique name: '{candidate_name}'")
            return candidate_name
        else:
            print(f"    Unavailable: {message}")
    
    raise Exception(f"Could not generate unique name after {max_attempts} attempts")

# Generate unique names for different environments
environments = ["dev", "test", "staging", "prod"]
project_name = "smartfactory"

print("Generating Environment-Specific IoT Hub Names:")
print("=" * 55)

environment_hubs = {}
for env in environments:
    try:
        base_name = f"{project_name}-{env}"
        unique_name = generate_unique_hub_name(base_name)
        environment_hubs[env] = unique_name
        print(f"✓ {env.upper()}: {unique_name}")
    except Exception as e:
        print(f"✗ {env.upper()}: {e}")
    print()

# Display deployment-ready names
if environment_hubs:
    print("Deployment-Ready IoT Hub Names:")
    print("=" * 40)
    for env, name in environment_hubs.items():
        print(f"  {env.upper()}_IOT_HUB_NAME='{name}'")

Discovering IoT Hub API capabilities

def discover_api_operations():
    """Discover and categorize all available IoT Hub API operations."""
    
    print("Discovering IoT Hub API Operations:")
    print("=" * 40)
    
    try:
        # Get all available operations
        operations = list(client.operations.list())
        
        print(f"Total API Operations: {len(operations)}")
        print()
        
        # Categorize operations by resource type
        operation_categories = {}
        
        for operation in operations:
            # Extract resource type from operation name
            if operation.name:
                parts = operation.name.split('/')
                if len(parts) >= 2:
                    provider = parts[0]  # e.g., "Microsoft.Devices"
                    resource_type = parts[1]  # e.g., "IotHubs"
                    
                    if resource_type not in operation_categories:
                        operation_categories[resource_type] = []
                    
                    operation_categories[resource_type].append(operation)
                else:
                    # Handle operations without clear categorization
                    if "Other" not in operation_categories:
                        operation_categories["Other"] = []
                    operation_categories["Other"].append(operation)
        
        # Display categorized operations
        for category, ops in sorted(operation_categories.items()):
            print(f"{category} Operations ({len(ops)} total):")
            print("-" * (len(category) + 20))
            
            for op in sorted(ops, key=lambda x: x.name):
                print(f"  Operation: {op.name}")
                if op.display and op.display.description:
                    print(f"    Description: {op.display.description}")
                if op.display and op.display.operation:
                    print(f"    Display Name: {op.display.operation}")
                if hasattr(op, 'origin') and op.origin:
                    print(f"    Origin: {op.origin}")
                print()
        
        return operation_categories
        
    except Exception as e:
        print(f"Failed to discover API operations: {e}")
        return {}

# Discover available operations
api_operations = discover_api_operations()

# Generate API operation summary
if api_operations:
    print("API Operation Summary:")
    print("=" * 25)
    
    total_operations = sum(len(ops) for ops in api_operations.values())
    print(f"Total Operations: {total_operations}")
    
    for category, ops in sorted(api_operations.items(), key=lambda x: len(x[1]), reverse=True):
        print(f"  {category}: {len(ops)} operations")

Building deployment validation pipeline

import json
from typing import List, Dict, Any

def create_deployment_validation_report(hub_names: List[str], target_regions: List[str]) -> Dict[str, Any]:
    """
    Create comprehensive deployment validation report for multiple IoT Hubs.
    
    Args:
        hub_names: List of proposed IoT Hub names
        target_regions: List of target Azure regions
        
    Returns:
        Validation report with availability and recommendations
    """
    
    validation_report = {
        "timestamp": datetime.utcnow().isoformat(),
        "validation_type": "pre_deployment",
        "requested_hubs": hub_names,
        "target_regions": target_regions,
        "name_validation": {},
        "api_capabilities": {},
        "recommendations": [],
        "deployment_ready": False
    }
    
    print("Creating Deployment Validation Report...")
    print("=" * 45)
    
    # Validate each hub name
    print("1. Validating IoT Hub names...")
    available_names = []
    unavailable_names = []
    
    for name in hub_names:
        print(f"   Checking: {name}")
        is_available, message = validate_hub_name(name)
        
        validation_report["name_validation"][name] = {
            "available": is_available,
            "message": message
        }
        
        if is_available:
            available_names.append(name)
            print(f"     ✓ Available")
        else:
            unavailable_names.append(name)
            print(f"     ✗ {message}")
    
    # API capability check
    print("\n2. Checking API capabilities...")
    try:
        operations = list(client.operations.list())
        validation_report["api_capabilities"] = {
            "total_operations": len(operations),
            "client_accessible": True,
            "key_operations_available": True
        }
        print(f"   ✓ {len(operations)} API operations available")
    except Exception as e:
        validation_report["api_capabilities"] = {
            "total_operations": 0,
            "client_accessible": False,
            "error": str(e)
        }
        print(f"   ✗ API access failed: {e}")
    
    # Generate recommendations
    print("\n3. Generating recommendations...")
    
    if unavailable_names:
        validation_report["recommendations"].append({
            "type": "name_availability",
            "severity": "error",
            "message": f"The following names are unavailable: {', '.join(unavailable_names)}",
            "action": "Choose alternative names or modify existing names"
        })
    
    if len(available_names) == 0:
        validation_report["recommendations"].append({
            "type": "deployment_blocker",
            "severity": "critical", 
            "message": "No valid IoT Hub names available for deployment",
            "action": "Generate new unique names before proceeding"
        })
    
    if len(target_regions) > 1:
        validation_report["recommendations"].append({
            "type": "multi_region",
            "severity": "info",
            "message": f"Deployment planned for {len(target_regions)} regions",
            "action": "Ensure consistent naming and configuration across regions"
        })
    
    # Determine deployment readiness
    validation_report["deployment_ready"] = (
        len(available_names) > 0 and 
        validation_report["api_capabilities"].get("client_accessible", False)
    )
    
    # Summary
    print("\n4. Validation Summary:")
    print(f"   Available names: {len(available_names)}")
    print(f"   Unavailable names: {len(unavailable_names)}")
    print(f"   API accessible: {validation_report['api_capabilities'].get('client_accessible', False)}")
    print(f"   Deployment ready: {validation_report['deployment_ready']}")
    
    return validation_report

# Create validation report for deployment pipeline
deployment_names = [
    "smartfactory-prod-hub",
    "smartfactory-dev-hub", 
    "smartfactory-test-hub"
]

deployment_regions = [
    "East US",
    "West US 2",
    "North Europe"
]

validation_report = create_deployment_validation_report(deployment_names, deployment_regions)

# Save validation report
report_filename = f"iot_hub_deployment_validation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
with open(report_filename, "w") as f:
    json.dump(validation_report, f, indent=2, default=str)

print(f"\nValidation report saved to: {report_filename}")

# Display final status
if validation_report["deployment_ready"]:
    print("\n✅ DEPLOYMENT VALIDATION PASSED")
    print("   Ready to proceed with IoT Hub deployment")
else:
    print("\n❌ DEPLOYMENT VALIDATION FAILED")
    print("   Address issues before deployment:")
    for rec in validation_report["recommendations"]:
        if rec["severity"] in ["error", "critical"]:
            print(f"   - {rec['message']}")

Types

OperationInputs

Request model for name availability validation containing the proposed IoT Hub name and validation parameters.

class OperationInputs:
    """Name availability check request."""
    name: str  # Proposed IoT Hub name to validate for availability

IotHubNameAvailabilityInfo

Response model containing name availability status and detailed information about availability constraints or conflicts.

class IotHubNameAvailabilityInfo:
    """IoT Hub name availability result."""
    name_available: bool    # Whether the name is available for use
    reason: str             # Reason for unavailability (if applicable)
    message: str            # Detailed message about availability status

Operation

API operation metadata including name, description, and capability information for service discovery and documentation.

class Operation:
    """API operation details."""
    name: str                    # Operation name (e.g., "Microsoft.Devices/IotHubs/read")
    display: OperationDisplay    # Human-readable operation information
    origin: str                  # Origin of the operation (user, system)

OperationDisplay

Human-readable operation information including description and display names for API documentation and tooling.

class OperationDisplay:
    """Operation display information."""
    provider: str       # Resource provider name
    resource: str       # Resource type name
    operation: str      # Operation display name
    description: str    # Detailed operation description

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