0
# Utility Operations
1
2
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.
3
4
## Capabilities
5
6
### Name Availability Checking
7
8
Validate IoT Hub name availability before resource creation to prevent deployment failures and ensure compliance with Azure naming requirements and global uniqueness constraints.
9
10
```python { .api }
11
def check_name_availability(operation_inputs: OperationInputs, **kwargs) -> IotHubNameAvailabilityInfo:
12
"""
13
Check if an IoT Hub name is available for use in Azure.
14
15
Args:
16
operation_inputs: Name validation request containing the proposed hub name
17
18
Returns:
19
IotHubNameAvailabilityInfo: Availability status and detailed reason if unavailable
20
"""
21
```
22
23
### API Operation Discovery
24
25
Discover all available REST API operations for the IoT Hub service, useful for understanding capabilities, building custom tooling, and API exploration.
26
27
```python { .api }
28
def list() -> ItemPaged[Operation]:
29
"""
30
List all available IoT Hub management REST API operations.
31
32
Returns:
33
ItemPaged[Operation]: Complete list of API operations with metadata and descriptions
34
"""
35
```
36
37
## Usage Examples
38
39
### Validating IoT Hub names before deployment
40
41
```python
42
from azure.identity import DefaultAzureCredential
43
from azure.mgmt.iothub import IotHubClient
44
from azure.mgmt.iothub.models import OperationInputs
45
46
# Initialize client
47
credential = DefaultAzureCredential()
48
client = IotHubClient(credential, "subscription-id")
49
50
def validate_hub_name(proposed_name: str) -> tuple[bool, str]:
51
"""
52
Validate if an IoT Hub name is available and meets requirements.
53
54
Returns:
55
Tuple of (is_available, message)
56
"""
57
58
print(f"Checking availability for IoT Hub name: '{proposed_name}'")
59
60
try:
61
# Create validation request
62
name_request = OperationInputs(name=proposed_name)
63
64
# Check name availability
65
availability_result = client.iot_hub_resource.check_name_availability(name_request)
66
67
print(f" Name Available: {availability_result.name_available}")
68
69
if availability_result.name_available:
70
print(f" ✓ '{proposed_name}' is available for use")
71
return True, "Name is available"
72
else:
73
print(f" ✗ '{proposed_name}' is not available")
74
print(f" Reason: {availability_result.reason}")
75
print(f" Message: {availability_result.message}")
76
return False, f"{availability_result.reason}: {availability_result.message}"
77
78
except Exception as e:
79
error_msg = f"Failed to check name availability: {e}"
80
print(f" ✗ {error_msg}")
81
return False, error_msg
82
83
# Test various hub names
84
test_names = [
85
"my-iot-hub-prod",
86
"contoso-manufacturing-hub",
87
"iot-hub-dev-001",
88
"azure-iot-hub", # Likely taken
89
"my-unique-hub-12345",
90
"test" # Too short
91
]
92
93
print("IoT Hub Name Validation Results:")
94
print("=" * 45)
95
96
available_names = []
97
unavailable_names = []
98
99
for name in test_names:
100
is_available, message = validate_hub_name(name)
101
102
if is_available:
103
available_names.append(name)
104
else:
105
unavailable_names.append((name, message))
106
107
print() # Add spacing between checks
108
109
# Summary
110
print("Summary:")
111
print(f" Available names: {len(available_names)}")
112
print(f" Unavailable names: {len(unavailable_names)}")
113
114
if available_names:
115
print("\nAvailable for use:")
116
for name in available_names:
117
print(f" ✓ {name}")
118
119
if unavailable_names:
120
print("\nNot available:")
121
for name, reason in unavailable_names:
122
print(f" ✗ {name} - {reason}")
123
```
124
125
### Generating unique IoT Hub names
126
127
```python
128
import random
129
import string
130
from datetime import datetime
131
132
def generate_unique_hub_name(base_name: str, max_attempts: int = 10) -> str:
133
"""
134
Generate a unique IoT Hub name by appending random suffixes.
135
136
Args:
137
base_name: Base name for the IoT Hub
138
max_attempts: Maximum number of generation attempts
139
140
Returns:
141
Available IoT Hub name, or raises exception if none found
142
"""
143
144
print(f"Generating unique IoT Hub name based on: '{base_name}'")
145
146
for attempt in range(max_attempts):
147
# Generate suffix with random characters
148
suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
149
candidate_name = f"{base_name}-{suffix}"
150
151
# Ensure name meets length requirements (3-50 characters)
152
if len(candidate_name) > 50:
153
candidate_name = candidate_name[:50]
154
elif len(candidate_name) < 3:
155
candidate_name = candidate_name + "001"
156
157
print(f" Attempt {attempt + 1}: '{candidate_name}'")
158
159
is_available, message = validate_hub_name(candidate_name)
160
161
if is_available:
162
print(f" ✓ Generated unique name: '{candidate_name}'")
163
return candidate_name
164
else:
165
print(f" Unavailable: {message}")
166
167
raise Exception(f"Could not generate unique name after {max_attempts} attempts")
168
169
# Generate unique names for different environments
170
environments = ["dev", "test", "staging", "prod"]
171
project_name = "smartfactory"
172
173
print("Generating Environment-Specific IoT Hub Names:")
174
print("=" * 55)
175
176
environment_hubs = {}
177
for env in environments:
178
try:
179
base_name = f"{project_name}-{env}"
180
unique_name = generate_unique_hub_name(base_name)
181
environment_hubs[env] = unique_name
182
print(f"✓ {env.upper()}: {unique_name}")
183
except Exception as e:
184
print(f"✗ {env.upper()}: {e}")
185
print()
186
187
# Display deployment-ready names
188
if environment_hubs:
189
print("Deployment-Ready IoT Hub Names:")
190
print("=" * 40)
191
for env, name in environment_hubs.items():
192
print(f" {env.upper()}_IOT_HUB_NAME='{name}'")
193
```
194
195
### Discovering IoT Hub API capabilities
196
197
```python
198
def discover_api_operations():
199
"""Discover and categorize all available IoT Hub API operations."""
200
201
print("Discovering IoT Hub API Operations:")
202
print("=" * 40)
203
204
try:
205
# Get all available operations
206
operations = list(client.operations.list())
207
208
print(f"Total API Operations: {len(operations)}")
209
print()
210
211
# Categorize operations by resource type
212
operation_categories = {}
213
214
for operation in operations:
215
# Extract resource type from operation name
216
if operation.name:
217
parts = operation.name.split('/')
218
if len(parts) >= 2:
219
provider = parts[0] # e.g., "Microsoft.Devices"
220
resource_type = parts[1] # e.g., "IotHubs"
221
222
if resource_type not in operation_categories:
223
operation_categories[resource_type] = []
224
225
operation_categories[resource_type].append(operation)
226
else:
227
# Handle operations without clear categorization
228
if "Other" not in operation_categories:
229
operation_categories["Other"] = []
230
operation_categories["Other"].append(operation)
231
232
# Display categorized operations
233
for category, ops in sorted(operation_categories.items()):
234
print(f"{category} Operations ({len(ops)} total):")
235
print("-" * (len(category) + 20))
236
237
for op in sorted(ops, key=lambda x: x.name):
238
print(f" Operation: {op.name}")
239
if op.display and op.display.description:
240
print(f" Description: {op.display.description}")
241
if op.display and op.display.operation:
242
print(f" Display Name: {op.display.operation}")
243
if hasattr(op, 'origin') and op.origin:
244
print(f" Origin: {op.origin}")
245
print()
246
247
return operation_categories
248
249
except Exception as e:
250
print(f"Failed to discover API operations: {e}")
251
return {}
252
253
# Discover available operations
254
api_operations = discover_api_operations()
255
256
# Generate API operation summary
257
if api_operations:
258
print("API Operation Summary:")
259
print("=" * 25)
260
261
total_operations = sum(len(ops) for ops in api_operations.values())
262
print(f"Total Operations: {total_operations}")
263
264
for category, ops in sorted(api_operations.items(), key=lambda x: len(x[1]), reverse=True):
265
print(f" {category}: {len(ops)} operations")
266
```
267
268
### Building deployment validation pipeline
269
270
```python
271
import json
272
from typing import List, Dict, Any
273
274
def create_deployment_validation_report(hub_names: List[str], target_regions: List[str]) -> Dict[str, Any]:
275
"""
276
Create comprehensive deployment validation report for multiple IoT Hubs.
277
278
Args:
279
hub_names: List of proposed IoT Hub names
280
target_regions: List of target Azure regions
281
282
Returns:
283
Validation report with availability and recommendations
284
"""
285
286
validation_report = {
287
"timestamp": datetime.utcnow().isoformat(),
288
"validation_type": "pre_deployment",
289
"requested_hubs": hub_names,
290
"target_regions": target_regions,
291
"name_validation": {},
292
"api_capabilities": {},
293
"recommendations": [],
294
"deployment_ready": False
295
}
296
297
print("Creating Deployment Validation Report...")
298
print("=" * 45)
299
300
# Validate each hub name
301
print("1. Validating IoT Hub names...")
302
available_names = []
303
unavailable_names = []
304
305
for name in hub_names:
306
print(f" Checking: {name}")
307
is_available, message = validate_hub_name(name)
308
309
validation_report["name_validation"][name] = {
310
"available": is_available,
311
"message": message
312
}
313
314
if is_available:
315
available_names.append(name)
316
print(f" ✓ Available")
317
else:
318
unavailable_names.append(name)
319
print(f" ✗ {message}")
320
321
# API capability check
322
print("\n2. Checking API capabilities...")
323
try:
324
operations = list(client.operations.list())
325
validation_report["api_capabilities"] = {
326
"total_operations": len(operations),
327
"client_accessible": True,
328
"key_operations_available": True
329
}
330
print(f" ✓ {len(operations)} API operations available")
331
except Exception as e:
332
validation_report["api_capabilities"] = {
333
"total_operations": 0,
334
"client_accessible": False,
335
"error": str(e)
336
}
337
print(f" ✗ API access failed: {e}")
338
339
# Generate recommendations
340
print("\n3. Generating recommendations...")
341
342
if unavailable_names:
343
validation_report["recommendations"].append({
344
"type": "name_availability",
345
"severity": "error",
346
"message": f"The following names are unavailable: {', '.join(unavailable_names)}",
347
"action": "Choose alternative names or modify existing names"
348
})
349
350
if len(available_names) == 0:
351
validation_report["recommendations"].append({
352
"type": "deployment_blocker",
353
"severity": "critical",
354
"message": "No valid IoT Hub names available for deployment",
355
"action": "Generate new unique names before proceeding"
356
})
357
358
if len(target_regions) > 1:
359
validation_report["recommendations"].append({
360
"type": "multi_region",
361
"severity": "info",
362
"message": f"Deployment planned for {len(target_regions)} regions",
363
"action": "Ensure consistent naming and configuration across regions"
364
})
365
366
# Determine deployment readiness
367
validation_report["deployment_ready"] = (
368
len(available_names) > 0 and
369
validation_report["api_capabilities"].get("client_accessible", False)
370
)
371
372
# Summary
373
print("\n4. Validation Summary:")
374
print(f" Available names: {len(available_names)}")
375
print(f" Unavailable names: {len(unavailable_names)}")
376
print(f" API accessible: {validation_report['api_capabilities'].get('client_accessible', False)}")
377
print(f" Deployment ready: {validation_report['deployment_ready']}")
378
379
return validation_report
380
381
# Create validation report for deployment pipeline
382
deployment_names = [
383
"smartfactory-prod-hub",
384
"smartfactory-dev-hub",
385
"smartfactory-test-hub"
386
]
387
388
deployment_regions = [
389
"East US",
390
"West US 2",
391
"North Europe"
392
]
393
394
validation_report = create_deployment_validation_report(deployment_names, deployment_regions)
395
396
# Save validation report
397
report_filename = f"iot_hub_deployment_validation_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
398
with open(report_filename, "w") as f:
399
json.dump(validation_report, f, indent=2, default=str)
400
401
print(f"\nValidation report saved to: {report_filename}")
402
403
# Display final status
404
if validation_report["deployment_ready"]:
405
print("\n✅ DEPLOYMENT VALIDATION PASSED")
406
print(" Ready to proceed with IoT Hub deployment")
407
else:
408
print("\n❌ DEPLOYMENT VALIDATION FAILED")
409
print(" Address issues before deployment:")
410
for rec in validation_report["recommendations"]:
411
if rec["severity"] in ["error", "critical"]:
412
print(f" - {rec['message']}")
413
```
414
415
## Types
416
417
### OperationInputs
418
Request model for name availability validation containing the proposed IoT Hub name and validation parameters.
419
420
```python
421
class OperationInputs:
422
"""Name availability check request."""
423
name: str # Proposed IoT Hub name to validate for availability
424
```
425
426
### IotHubNameAvailabilityInfo
427
Response model containing name availability status and detailed information about availability constraints or conflicts.
428
429
```python
430
class IotHubNameAvailabilityInfo:
431
"""IoT Hub name availability result."""
432
name_available: bool # Whether the name is available for use
433
reason: str # Reason for unavailability (if applicable)
434
message: str # Detailed message about availability status
435
```
436
437
### Operation
438
API operation metadata including name, description, and capability information for service discovery and documentation.
439
440
```python
441
class Operation:
442
"""API operation details."""
443
name: str # Operation name (e.g., "Microsoft.Devices/IotHubs/read")
444
display: OperationDisplay # Human-readable operation information
445
origin: str # Origin of the operation (user, system)
446
```
447
448
### OperationDisplay
449
Human-readable operation information including description and display names for API documentation and tooling.
450
451
```python
452
class OperationDisplay:
453
"""Operation display information."""
454
provider: str # Resource provider name
455
resource: str # Resource type name
456
operation: str # Operation display name
457
description: str # Detailed operation description
458
```