Microsoft Azure IoT Hub Management Client Library for programmatic management of Azure IoT Hub resources through the Azure Resource Manager API
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
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.
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
"""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
"""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}")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}'")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")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']}")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 availabilityResponse 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 statusAPI 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)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 descriptionInstall with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-iothub