in azure/multiapi/storagev2/blob/v2022_11_02/_shared/response_handlers.py [0:0]
def process_storage_error(storage_error): # pylint:disable=too-many-statements
raise_error = HttpResponseError
serialized = False
if not storage_error.response:
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:
error_body = error_body or 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
if error_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:
raise error