def _create_or_update_custom_location()

in azext_edge/edge/providers/orchestration/work.py [0:0]

• 26 lines of code
• 3 McCabe index (conditional complexity)

    def _create_or_update_custom_location(self, extension_ids: Iterable[str]) -> dict:
        try:
            return self.custom_locations.create(
                name=self._targets.custom_location_name,
                resource_group_name=self._targets.resource_group_name,
                host_resource_id=self._resource_map.connected_cluster.resource_id,
                namespace=self._targets.cluster_namespace,
                display_name=self._targets.custom_location_name,
                location=self._targets.location,
                cluster_extension_ids=extension_ids,
                tags=self._targets.tags,
            )
        except HttpResponseError as http_exc:
            if http_exc.error.code == "UnauthorizedNamespaceError":
                explain = (
                    "[IoT Ops explanation]\n\nThis error generally happens for two reasons.\n"
                    "- The arc custom locations feature was not enabled.\n"
                    "- The arc custom locations feature was not enabled with the correct OID.\n\n"
                    "To resolve the issue, re-run create after applying the instructions at the aka.ms "
                    "link provided in the error."
                )
                cl_error_prefix = "Custom Locations Error:\n"
                raise ValidationError(
                    f"{insert_newlines(f'{cl_error_prefix}{http_exc.error.message}', 140)}\n\n{explain}"
                )
            raise http_exc