azure-iot-hub

import os from azure.iot.hub import IoTHubRegistryManager # Retrieve IoT Hub connection string from environment variables for security IOT_HUB_CONNECTION_STRING = os.environ.get("AZURE_IOT_HUB_CONNECTION_STRING", "") if not IOT_HUB_CONNECTION_STRING: raise ValueError("AZURE_IOT_HUB_CONNECTION_STRING environment variable not set.") DEVICE_ID = "myPythonManagedDevice" manager = None try: # Create an instance of the IoTHubRegistryManager # This connection string should be for a service/iothubowner policy manager = IoTHubRegistryManager.from_connection_string(IOT_HUB_CONNECTION_STRING) print(f"Attempting to create device: {DEVICE_ID}") # Create a new device identity # If the device already exists, this will raise an IoTHubError (409 Conflict) try: device = manager.create_device(DEVICE_ID) print(f"Device '{device.device_id}' created successfully.") except Exception as e: if "409001 DeviceAlreadyExists" in str(e): print(f"Device '{DEVICE_ID}' already exists. Retrieving it.") device = manager.get_device(DEVICE_ID) print(f"Retrieved device '{device.device_id}'.") else: raise e # You can inspect the device object, e.g., its primary key if device.authentication and device.authentication.symmetric_key: print(f"Primary key: {device.authentication.symmetric_key.primary_key}") print("Successfully interacted with IoT Hub Registry.") finally: # Close the manager to release resources if manager: print("Closing IoTHubRegistryManager...") manager.close() print("Manager closed.")