def _apply_sr_role_assignment()

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

• 29 lines of code
• 4 McCabe index (conditional complexity)

    def _apply_sr_role_assignment(self) -> Optional[str]:
        ops_ext = self.ops_extension
        if not ops_ext:
            raise ValidationError("IoT Operations extension not detected. Please run 'az iot ops create'.")
        # TODO - add non-success provisioningState
        ops_ext_principal_id = ops_ext.get("identity", {}).get("principalId")
        if not ops_ext_principal_id:
            raise ValidationError(
                "Unable to determine the IoT Operations system-managed identity principal Id.\n"
                "Please re-deploy via 'az iot ops create'."
            )

        try:
            schema_registry_id_parts = parse_resource_id(self._targets.schema_registry_resource_id)
            self.permission_manager.apply_role_assignment(
                scope=self._targets.schema_registry_resource_id,
                principal_id=ops_ext_principal_id,
                role_def_id=ROLE_DEF_FORMAT_STR.format(
                    subscription_id=schema_registry_id_parts.subscription_id,
                    role_id=CONTRIBUTOR_ROLE_ID,
                ),
                principal_type=PrincipalType.SERVICE_PRINCIPAL.value,
            )
        except HttpResponseError as e:
            self._warnings.append(
                get_user_msg_warn_ra(
                    prefix=f"Role assignment failed with:\n{str(e)}",
                    principal_id=ops_ext_principal_id,
                    scope=self._targets.schema_registry_resource_id,
                )
            )