def process_storage_error()

in azure/multiapi/storagev2/fileshare/v2025_05_05/_shared/response_handlers.py [0:0]

• 88 lines of code
• 31 McCabe index (conditional complexity)

def process_storage_error(storage_error) -> NoReturn: # type: ignore [misc] # pylint:disable=too-many-statements, too-many-branches
    raise_error = HttpResponseError
    serialized = False
    if isinstance(storage_error, AzureSigningError):
        storage_error.message = storage_error.message + \
            '. This is likely due to an invalid shared key. Please check your shared key and try again.'
    if not storage_error.response or storage_error.response.status_code in [200, 204]:
        raise storage_error
    # If it is one of those three then it has been serialized prior by the generated layer.
    if isinstance(storage_error, (PartialBatchErrorException,
                                  ClientAuthenticationError, ResourceNotFoundError, ResourceExistsError)):
        serialized = True
    error_code = storage_error.response.headers.get('x-ms-error-code')
    error_message = storage_error.message
    additional_data = {}
    error_dict = {}
    try:
        error_body = ContentDecodePolicy.deserialize_from_http_generics(storage_error.response)
        try:
            if error_body is None or len(error_body) == 0:
                error_body = storage_error.response.reason
        except AttributeError:
            error_body = ''
        # If it is an XML response
        if isinstance(error_body, Element):
            error_dict = {
                child.tag.lower(): child.text
                for child in error_body
            }
        # If it is a JSON response
        elif isinstance(error_body, dict):
            error_dict = error_body.get('error', {})
        elif not error_code:
            _LOGGER.warning(
                'Unexpected return type %s from ContentDecodePolicy.deserialize_from_http_generics.', type(error_body))
            error_dict = {'message': str(error_body)}

        # If we extracted from a Json or XML response
        # There is a chance error_dict is just a string
        if error_dict and isinstance(error_dict, dict):
            error_code = error_dict.get('code')
            error_message = error_dict.get('message')
            additional_data = {k: v for k, v in error_dict.items() if k not in {'code', 'message'}}
    except DecodeError:
        pass

    try:
        # This check would be unnecessary if we have already serialized the error
        if error_code and not serialized:
            error_code = StorageErrorCode(error_code)
            if error_code in [StorageErrorCode.condition_not_met,
                              StorageErrorCode.blob_overwritten]:
                raise_error = ResourceModifiedError
            if error_code in [StorageErrorCode.invalid_authentication_info,
                              StorageErrorCode.authentication_failed]:
                raise_error = ClientAuthenticationError
            if error_code in [StorageErrorCode.resource_not_found,
                              StorageErrorCode.cannot_verify_copy_source,
                              StorageErrorCode.blob_not_found,
                              StorageErrorCode.queue_not_found,
                              StorageErrorCode.container_not_found,
                              StorageErrorCode.parent_not_found,
                              StorageErrorCode.share_not_found]:
                raise_error = ResourceNotFoundError
            if error_code in [StorageErrorCode.account_already_exists,
                              StorageErrorCode.account_being_created,
                              StorageErrorCode.resource_already_exists,
                              StorageErrorCode.resource_type_mismatch,
                              StorageErrorCode.blob_already_exists,
                              StorageErrorCode.queue_already_exists,
                              StorageErrorCode.container_already_exists,
                              StorageErrorCode.container_being_deleted,
                              StorageErrorCode.queue_being_deleted,
                              StorageErrorCode.share_already_exists,
                              StorageErrorCode.share_being_deleted]:
                raise_error = ResourceExistsError
    except ValueError:
        # Got an unknown error code
        pass

    # Error message should include all the error properties
    try:
        error_message += f"\nErrorCode:{error_code.value}"
    except AttributeError:
        error_message += f"\nErrorCode:{error_code}"
    for name, info in additional_data.items():
        error_message += f"\n{name}:{info}"

    # No need to create an instance if it has already been serialized by the generated layer
    if serialized:
        storage_error.message = error_message
        error = storage_error
    else:
        error = raise_error(message=error_message, response=storage_error.response)
    # Ensure these properties are stored in the error instance as well (not just the error message)
    error.error_code = error_code
    error.additional_info = additional_data
    # error.args is what's surfaced on the traceback - show error message in all cases
    error.args = (error.message,)
    try:
        # `from None` prevents us from double printing the exception (suppresses generated layer error context)
        exec("raise error from None")   # pylint: disable=exec-used # nosec
    except SyntaxError as exc:
        raise error from exc