Replace deprecated CLIError with new error types

src/quantum/azext_quantum/_params.py

@@ -5,7 +5,8 @@
 # pylint: disable=line-too-long,protected-access
 
 import argparse
-from knack.arguments import CLIArgumentType, CLIError
+from knack.arguments import CLIArgumentType
+from azure.cli.core.azclierror import InvalidArgumentValueError
 
 
 class JobParamsAction(argparse._AppendAction):
@@ -19,8 +20,8 @@ def get_action(self, values, option_string):
             try:
                 key, value = item.split('=', 1)
                 params[key] = value
-            except ValueError:
-                raise CLIError('Usage error: {} KEY=VALUE [KEY=VALUE ...]'.format(option_string))
+            except ValueError as e:
+                raise InvalidArgumentValueError('Usage error: {} KEY=VALUE [KEY=VALUE ...]'.format(option_string)) from e
         return params
 
 

src/quantum/azext_quantum/operations/job.py

@@ -7,7 +7,8 @@
 
 import logging
 
-from knack.util import CLIError
+from azure.cli.core.azclierror import (FileOperationError, AzureInternalError,
+                                       InvalidArgumentValueError, AzureResponseError)
 
 from .._client_factory import cf_jobs, _get_data_credentials, base_url
 from .workspace import WorkspaceInfo
@@ -43,11 +44,11 @@ def _check_dotnet_available():
     try:
         import subprocess
         result = subprocess.run(args, stdout=subprocess.PIPE, check=False)
-    except FileNotFoundError:
-        raise CLIError(f"Could not find 'dotnet' on the system.")
+    except FileNotFoundError as e:
+        raise FileOperationError("Could not find 'dotnet' on the system.") from e
 
     if result.returncode != 0:
-        raise CLIError(f"Failed to run 'dotnet'. (Error {result.returncode})")
+        raise FileOperationError(f"Failed to run 'dotnet'. (Error {result.returncode})")
 
 
 def build(cmd, target_id=None, project=None):
@@ -78,7 +79,7 @@ def build(cmd, target_id=None, project=None):
     # If we got here, we might have encountered an error during compilation, so propagate standard output to the user.
     logger.error(f"Compilation stage failed with error code {result.returncode}")
     print(result.stdout.decode('ascii'))
-    raise CLIError("Failed to compile program.")
+    raise AzureInternalError("Failed to compile program.")
 
 
 def _generate_submit_args(program_args, ws, target, token, project, job_name, shots, storage, job_params):
@@ -190,7 +191,7 @@ def submit(cmd, program_args, resource_group_name=None, workspace_name=None, loc
     # The program compiled succesfully, but executing the stand-alone .exe failed to run.
     logger.error(f"Submission of job failed with error code {result.returncode}")
     print(result.stdout.decode('ascii'))
-    raise CLIError("Failed to submit job.")
+    raise AzureInternalError("Failed to submit job.")
 
 
 def _parse_blob_url(url):
@@ -202,8 +203,8 @@ def _parse_blob_url(url):
         container = o.path.split('/')[-2]
         blob = o.path.split('/')[-1]
         sas_token = o.query
-    except IndexError:
-        raise CLIError(f"Failed to parse malformed blob URL: {url}")
+    except IndexError as e:
+        raise InvalidArgumentValueError(f"Failed to parse malformed blob URL: {url}") from e
 
     return {
         "account_name": account_name,
@@ -253,7 +254,7 @@ def output(cmd, job_id, resource_group_name=None, workspace_name=None, location=
                 while result_start_line >= 0 and not lines[result_start_line].startswith('"'):
                     result_start_line -= 1
             if result_start_line < 0:
-                raise CLIError("Job output is malformed, mismatched quote characters.")
+                raise AzureResponseError("Job output is malformed, mismatched quote characters.")
 
             # Print the job output and then the result of the operation as a histogram.
             # If the result is a string, trim the quotation marks.

src/quantum/azext_quantum/operations/offerings.py

@@ -5,7 +5,7 @@
 
 # pylint: disable=line-too-long,redefined-builtin
 
-from knack.util import CLIError
+from azure.cli.core.azclierror import InvalidArgumentValueError, RequiredArgumentMissingError
 from .._client_factory import cf_offerings, cf_vm_image_term
 import time
 
@@ -42,7 +42,7 @@ def _get_publisher_and_offer_from_provider_id(providers, provider_id):
 
 def _valid_publisher_and_offer(provider, publisher, offer):
     if (offer is None or publisher is None):
-        raise CLIError(f"Provider '{provider}' not found.")
+        raise InvalidArgumentValueError(f"Provider '{provider}' not found.")
     if (offer == OFFER_NOT_AVAILABLE or publisher == PUBLISHER_NOT_AVAILABLE):
         # We show this information to the user to prevent a confusion when term commands take no effect.
         _show_info(f"No terms require to be accepted for provider '{provider}'.")
@@ -55,7 +55,7 @@ def list_offerings(cmd, location=None):
     Get the list of all provider offerings available on the given location.
     """
     if (not location):
-        raise CLIError("A location is required to list offerings available.")
+        raise RequiredArgumentMissingError("A location is required to list offerings available.")
     client = cf_offerings(cmd.cli_ctx)
     return client.list(location_name=location)
 

src/quantum/azext_quantum/operations/workspace.py

@@ -5,7 +5,8 @@
 
 # pylint: disable=line-too-long,redefined-builtin
 
-from knack.util import CLIError
+from azure.cli.core.azclierror import (InvalidArgumentValueError, AzureInternalError,
+                                       RequiredArgumentMissingError, ResourceNotFoundError)
 
 from .._client_factory import cf_workspaces, cf_quotas, cf_offerings
 from ..vendored_sdks.azure_mgmt_quantum.models import QuantumWorkspace
@@ -97,19 +98,19 @@ def _add_quantum_providers(cmd, workspace, providers):
     for pair in providers.split(','):
         es = [e.strip() for e in pair.split('/')]
         if (len(es) != 2):
-            raise CLIError(f"Invalid Provider/SKU specified: '{pair.strip()}'")
+            raise InvalidArgumentValueError(f"Invalid Provider/SKU specified: '{pair.strip()}'")
         provider_id = es[0]
         sku = es[1]
         (publisher, offer) = _get_publisher_and_offer_from_provider_id(providers_in_region, provider_id)
         if (offer is None or publisher is None):
-            raise CLIError(f"Provider '{provider_id}' not found in region {workspace.location}.")
+            raise InvalidArgumentValueError(f"Provider '{provider_id}' not found in region {workspace.location}.")
         providers_selected.append({'provider_id': provider_id, 'sku': sku, 'offer_id': offer, 'publisher_id': publisher})
     _show_tip(f"Workspace creation has been requested with the following providers:\n{providers_selected}")
     # Now that the providers have been requested, add each of them into the workspace
     for provider in providers_selected:
         if _provider_terms_need_acceptance(cmd, provider):
-            raise CLIError(f"Terms for Provider '{provider['provider_id']}' and SKU '{provider['sku']}' have not been accepted.\n"
-                           "Use command 'az quantum offerings accept-terms' to accept them.")
+            raise InvalidArgumentValueError(f"Terms for Provider '{provider['provider_id']}' and SKU '{provider['sku']}' have not been accepted.\n"
+                                            "Use command 'az quantum offerings accept-terms' to accept them.")
         p = Provider()
         p.provider_id = provider['provider_id']
         p.provider_sku = provider['sku']
@@ -123,7 +124,7 @@ def _create_role_assignment(cmd, quantum_workspace):
         try:
             create_role_assignment(cmd, role="Contributor", scope=quantum_workspace.storage_account, assignee=quantum_workspace.identity.principal_id)
             break
-        except (CloudError, CLIError) as e:
+        except (CloudError, AzureInternalError) as e:
             error = str(e.args).lower()
             if (("does not exist" in error) or ("cannot find" in error)):
                 print('.', end='', flush=True)
@@ -132,12 +133,12 @@ def _create_role_assignment(cmd, quantum_workspace):
                 continue
             raise e
         except Exception as x:
-            raise CLIError(f"Role assignment encountered exception ({type(x).__name__}): {x}")
+            raise AzureInternalError(f"Role assignment encountered exception ({type(x).__name__}): {x}")
     if (retry_attempts > 0):
         print()  # To end the line of the waiting indicators.
     if (retry_attempts == MAX_RETRIES_ROLE_ASSIGNMENT):
         max_time_in_seconds = MAX_RETRIES_ROLE_ASSIGNMENT * POLLING_TIME_DURATION
-        raise CLIError(f"Role assignment could not be added to storage account {quantum_workspace.storage_account} within {max_time_in_seconds} seconds.")
+        raise AzureInternalError(f"Role assignment could not be added to storage account {quantum_workspace.storage_account} within {max_time_in_seconds} seconds.")
     return quantum_workspace
 
 
@@ -147,16 +148,16 @@ def create(cmd, resource_group_name=None, workspace_name=None, location=None, st
     """
     client = cf_workspaces(cmd.cli_ctx)
     if (not workspace_name):
-        raise CLIError("An explicit workspace name is required for this command.")
+        raise RequiredArgumentMissingError("An explicit workspace name is required for this command.")
     if (not storage_account):
-        raise CLIError("A quantum workspace requires a valid storage account.")
+        raise RequiredArgumentMissingError("A quantum workspace requires a valid storage account.")
     if (not location):
-        raise CLIError("A location for the new quantum workspace is required.")
+        raise RequiredArgumentMissingError("A location for the new quantum workspace is required.")
     if (provider_sku_list is None):
-        raise CLIError("A list of Azure Quantum providers and SKUs is required.")
+        raise RequiredArgumentMissingError("A list of Azure Quantum providers and SKUs is required.")
     info = WorkspaceInfo(cmd, resource_group_name, workspace_name, location)
     if (not info.resource_group):
-        raise CLIError("Please run 'az quantum workspace set' first to select a default resource group.")
+        raise ResourceNotFoundError("Please run 'az quantum workspace set' first to select a default resource group.")
     quantum_workspace = _get_basic_quantum_workspace(location, info, storage_account)
     _add_quantum_providers(cmd, quantum_workspace, provider_sku_list)
     poller = client.begin_create_or_update(info.resource_group, info.name, quantum_workspace, polling=False)
@@ -175,7 +176,7 @@ def delete(cmd, resource_group_name=None, workspace_name=None):
     client = cf_workspaces(cmd.cli_ctx)
     info = WorkspaceInfo(cmd, resource_group_name, workspace_name)
     if (not info.resource_group) or (not info.name):
-        raise CLIError("Please run 'az quantum workspace set' first to select a default Quantum Workspace.")
+        raise ResourceNotFoundError("Please run 'az quantum workspace set' first to select a default Quantum Workspace.")
     client.begin_delete(info.resource_group, info.name, polling=False)
     # If we deleted the current workspace, clear it
     curr_ws = WorkspaceInfo(cmd)
@@ -202,7 +203,7 @@ def get(cmd, resource_group_name=None, workspace_name=None):
     client = cf_workspaces(cmd.cli_ctx)
     info = WorkspaceInfo(cmd, resource_group_name, workspace_name, None)
     if (not info.resource_group) or (not info.name):
-        raise CLIError("Please run 'az quantum workspace set' first to select a default Quantum Workspace.")
+        raise ResourceNotFoundError("Please run 'az quantum workspace set' first to select a default Quantum Workspace.")
     ws = client.get(info.resource_group, info.name)
     return ws