def download_devices()

in azext_iot/iothub/providers/state.py [0:0]

• 81 lines of code
• 25 McCabe index (conditional complexity)

    def download_devices(self, target: Dict[str, str]) -> dict:
        """
        Fetch devices and convert to the following structure:
        {
            "device_id": {
                "identity": { identity_properties (and properties shared with twin) },
                "twin" : { twin_properties },
                "parent" : parent_id,
                "modules" : {
                    "module_id" : {
                        "identity": { identity_properties },
                        "twin": { twin_properties }
                    }
                }
            }
        }
        """
        # if incorrect permissions, will fail to retrieve any devices
        devices = {}
        try:
            twins = _iot_device_twin_list(target=target, top=None)
        except AzCLIError:
            logger.warning(usr_msgs.SAVE_DEVICES_RETRIEVE_FAIL_MSG)
            return

        for i in tqdm(range(len(twins)), desc=usr_msgs.SAVE_DEVICE_DESC, ascii=" #"):
            device_twin = twins[i]
            device_id = device_twin["deviceId"]
            device_obj = {}

            if device_twin.get("parentScopes"):
                device_parent = device_twin["parentScopes"][0].split("://")[1]
                device_obj["parent"] = device_parent[:device_parent.rfind("-")]

            # Basic tier does not support device twins, modules
            if not device_twin.get("properties"):
                continue
            # put properties + tags into the saved twin
            device_twin["properties"].pop("reported")
            for key in ["$metadata", "$version"]:
                device_twin["properties"]["desired"].pop(key)

            device_obj["twin"] = {
                "properties": device_twin.pop("properties")
            }

            if device_twin.get("tags"):
                device_obj["twin"]["tags"] = device_twin.pop("tags")

            # create the device identity from the device twin
            # primary and secondary keys show up in the "show" output but not in the "list" output
            authentication = {
                "type": device_twin.pop("authenticationType"),
                "x509Thumbprint": device_twin.pop("x509Thumbprint")
            }
            if authentication["type"] == DeviceAuthApiType.sas.value:
                # Cannot retrieve the sas key for some reason - throw out the device
                try:
                    id2 = _iot_device_show(target=target, device_id=device_id)
                    authentication["symmetricKey"] = id2["authentication"]["symmetricKey"]
                except AzCLIError:
                    logger.warning(usr_msgs.SAVE_SPECIFIC_DEVICE_RETRIEVE_FAIL_MSG.format(device_id))
                    continue
            device_twin["authentication"] = authentication

            for key in IMMUTABLE_DEVICE_IDENTITY_FIELDS:
                device_twin.pop(key, None)
            device_obj["identity"] = device_twin

            # if unable to retrieve modules, log and continue without modules
            module_objs = {}
            try:
                module_objs = _iot_device_module_list(target=target, device_id=device_id)
            except AzCLIError:
                logger.warning(usr_msgs.SAVE_SPECIFIC_DEVICE_MODULES_RETRIEVE_FAIL_MSG.format(device_id))

            if module_objs:
                device_obj["modules"] = {}

            for module in module_objs:
                module = module.serialize()
                module_id = module["moduleId"]
                # Fail to retrieve module identity - log and continue without module
                try:
                    module_identity_show = _iot_device_module_show(
                        target=target, device_id=device_id, module_id=module_id
                    )
                except AzCLIError:
                    logger.warning(
                        usr_msgs.SAVE_SPECIFIC_DEVICE_SPECIFIC_MODULE_RETRIEVE_FAIL_MSG.format("identity", module_id, device_id)
                    )
                    continue
                module["authentication"] = module_identity_show["authentication"]

                for key in IMMUTABLE_MODULE_IDENTITY_FIELDS:
                    module.pop(key)

                # Fail to retrieve module twin - log and continue without module
                try:
                    module_twin = _iot_device_module_twin_show(
                        target=target, device_id=device_id, module_id=module_id
                    )
                except AzCLIError:
                    logger.warning(
                        usr_msgs.SAVE_SPECIFIC_DEVICE_SPECIFIC_MODULE_RETRIEVE_FAIL_MSG.format("twin", module_id, device_id)
                    )
                    continue

                for key in IMMUTABLE_AND_DUPLICATE_MODULE_TWIN_FIELDS:
                    module_twin.pop(key)
                for key in ["$metadata", "$version"]:
                    module_twin["properties"]["desired"].pop(key)
                module_twin["properties"].pop("reported")

                device_obj["modules"][module_id] = {
                    "identity": module,
                    "twin": module_twin
                }

            devices[device_id] = device_obj

        return devices