def _analyze_assets()

in azext_edge/edge/providers/orchestration/clone.py [0:0]
33 lines of code
3 McCabe index (conditional complexity)

    def _analyze_assets(self):
        nested_params = {
            **build_parameter(name=TemplateParams.CUSTOM_LOCATION_NAME.value),
        }
        instance_resource_id_expr = get_resource_id_by_param(
            "microsoft.iotoperations/instances", TemplateParams.INSTANCE_NAME
        )

        asset_endpoints = self.get_resources_of_type(resource_type="microsoft.deviceregistry/assetendpointprofiles")
        self._add_deployment(
            key=StateResourceKey.ASSET_ENDPOINT_PROFILE,
            api_version=REGISTRY_API_VERSION,
            data_iter=asset_endpoints,
            depends_on=[
                instance_resource_id_expr,
                get_resource_id_by_parts(
                    "Microsoft.Resources/deployments",
                    self.active_deployment[StateResourceKey.LISTENER][-1],
                ),
            ],
            parameters=nested_params,
        )

        # TODO: Should this not wait on AEP?
        assets = self.get_resources_of_type(resource_type="microsoft.deviceregistry/assets")
        if assets and asset_endpoints:
            self._add_deployment(
                key=StateResourceKey.ASSET,
                api_version=REGISTRY_API_VERSION,
                data_iter=assets,
                depends_on=get_resource_id_by_parts(
                    "Microsoft.Resources/deployments",
                    self.active_deployment[StateResourceKey.ASSET_ENDPOINT_PROFILE][-1],
                ),
                parameters=nested_params,
            )