# --------------------------------------------------------------------------
#
# Copyright (c) Microsoft Corporation. All rights reserved.
#
# The MIT License (MIT)
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the ""Software""), to
# deal in the Software without restriction, including without limitation the
# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
# sell copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED *AS IS*, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
# IN THE SOFTWARE.
#
# --------------------------------------------------------------------------
from __future__ import annotations
import json
import logging
import sys
from types import TracebackType
from typing import (
Callable,
Any,
Optional,
Union,
Type,
List,
Mapping,
TypeVar,
Generic,
Dict,
NoReturn,
TYPE_CHECKING,
)
from typing_extensions import Protocol, runtime_checkable
_LOGGER = logging.getLogger(__name__)
if TYPE_CHECKING:
from azure.core.pipeline.policies import RequestHistory
HTTPResponseType = TypeVar("HTTPResponseType")
HTTPRequestType = TypeVar("HTTPRequestType")
KeyType = TypeVar("KeyType")
ValueType = TypeVar("ValueType")
# To replace when typing.Self is available in our baseline
SelfODataV4Format = TypeVar("SelfODataV4Format", bound="ODataV4Format")
__all__ = [
"AzureError",
"ServiceRequestError",
"ServiceResponseError",
"HttpResponseError",
"DecodeError",
"ResourceExistsError",
"ResourceNotFoundError",
"ClientAuthenticationError",
"ResourceModifiedError",
"ResourceNotModifiedError",
"TooManyRedirectsError",
"ODataV4Format",
"ODataV4Error",
"StreamConsumedError",
"StreamClosedError",
"ResponseNotReadError",
"SerializationError",
"DeserializationError",
]
def raise_with_traceback(exception: Callable, *args: Any, message: str = "", **kwargs: Any) -> NoReturn:
"""Raise exception with a specified traceback.
This MUST be called inside a "except" clause.
.. note:: This method is deprecated since we don't support Python 2 anymore. Use raise/from instead.
:param Exception exception: Error type to be raised.
:param any args: Any additional args to be included with exception.
:keyword str message: Message to be associated with the exception. If omitted, defaults to an empty string.
"""
exc_type, exc_value, exc_traceback = sys.exc_info()
# If not called inside an "except", exc_type will be None. Assume it will not happen
if exc_type is None:
raise ValueError("raise_with_traceback can only be used in except clauses")
exc_msg = "{}, {}: {}".format(message, exc_type.__name__, exc_value)
error = exception(exc_msg, *args, **kwargs)
try:
raise error.with_traceback(exc_traceback) # pylint: disable=raise-missing-from
except AttributeError: # Python 2
error.__traceback__ = exc_traceback
raise error # pylint: disable=raise-missing-from
@runtime_checkable
class _HttpResponseCommonAPI(Protocol):
"""Protocol used by exceptions for HTTP response.
As HttpResponseError uses very few properties of HttpResponse, a protocol
is faster and simpler than import all the possible types (at least 6).
"""
@property
def reason(self) -> Optional[str]:
...
@property
def status_code(self) -> Optional[int]:
...
def text(self) -> str:
...
@property
def request(self) -> object: # object as type, since all we need is str() on it
...
class ErrorMap(Generic[KeyType, ValueType]):
"""Error Map class. To be used in map_error method, behaves like a dictionary.
It returns the error type if it is found in custom_error_map. Or return default_error
:param dict custom_error_map: User-defined error map, it is used to map status codes to error types.
:keyword error default_error: Default error type. It is returned if the status code is not found in custom_error_map
"""
def __init__(
self, # pylint: disable=unused-argument
custom_error_map: Optional[Mapping[KeyType, ValueType]] = None,
*,
default_error: Optional[ValueType] = None,
**kwargs: Any,
) -> None:
self._custom_error_map = custom_error_map or {}
self._default_error = default_error
def get(self, key: KeyType) -> Optional[ValueType]:
ret = self._custom_error_map.get(key)
if ret:
return ret
return self._default_error
def map_error(
status_code: int, response: _HttpResponseCommonAPI, error_map: Mapping[int, Type[HttpResponseError]]
) -> None:
if not error_map:
return
error_type = error_map.get(status_code)
if not error_type:
return
error = error_type(response=response)
raise error
[docs]class AzureError(Exception):
"""Base exception for all errors.
:param object message: The message object stringified as 'message' attribute
:keyword error: The original exception if any
:paramtype error: Exception
:ivar inner_exception: The exception passed with the 'error' kwarg
:vartype inner_exception: Exception
:ivar exc_type: The exc_type from sys.exc_info()
:ivar exc_value: The exc_value from sys.exc_info()
:ivar exc_traceback: The exc_traceback from sys.exc_info()
:ivar exc_msg: A string formatting of message parameter, exc_type and exc_value
:ivar str message: A stringified version of the message parameter
:ivar str continuation_token: A token reference to continue an incomplete operation. This value is optional
and will be `None` where continuation is either unavailable or not applicable.
"""
def __init__(self, message: Optional[object], *args: Any, **kwargs: Any) -> None:
self.inner_exception: Optional[BaseException] = kwargs.get("error")
exc_info = sys.exc_info()
self.exc_type: Optional[Type[Any]] = exc_info[0]
self.exc_value: Optional[BaseException] = exc_info[1]
self.exc_traceback: Optional[TracebackType] = exc_info[2]
self.exc_type = self.exc_type if self.exc_type else type(self.inner_exception)
self.exc_msg: str = "{}, {}: {}".format(message, self.exc_type.__name__, self.exc_value)
self.message: str = str(message)
self.continuation_token: Optional[str] = kwargs.get("continuation_token")
super(AzureError, self).__init__(self.message, *args)
[docs] def raise_with_traceback(self) -> None:
"""Raise the exception with the existing traceback.
.. deprecated:: 1.22.0
This method is deprecated as we don't support Python 2 anymore. Use raise/from instead.
"""
try:
raise super(AzureError, self).with_traceback(self.exc_traceback) # pylint: disable=raise-missing-from
except AttributeError:
self.__traceback__: Optional[TracebackType] = self.exc_traceback
raise self # pylint: disable=raise-missing-from
[docs]class ServiceRequestError(AzureError):
"""An error occurred while attempt to make a request to the service.
No request was sent.
"""
[docs]class ServiceResponseError(AzureError):
"""The request was sent, but the client failed to understand the response.
The connection may have timed out. These errors can be retried for idempotent or
safe operations"""
class ServiceRequestTimeoutError(ServiceRequestError):
"""Error raised when timeout happens"""
class ServiceResponseTimeoutError(ServiceResponseError):
"""Error raised when timeout happens"""
[docs]class HttpResponseError(AzureError):
"""A request was made, and a non-success status code was received from the service.
:param object message: The message object stringified as 'message' attribute
:param response: The response that triggered the exception.
:type response: ~azure.core.pipeline.transport.HttpResponse or ~azure.core.pipeline.transport.AsyncHttpResponse
:ivar reason: The HTTP response reason
:vartype reason: str
:ivar status_code: HttpResponse's status code
:vartype status_code: int
:ivar response: The response that triggered the exception.
:vartype response: ~azure.core.pipeline.transport.HttpResponse or ~azure.core.pipeline.transport.AsyncHttpResponse
:ivar model: The request body/response body model
:vartype model: ~msrest.serialization.Model
:ivar error: The formatted error
:vartype error: ODataV4Format
"""
def __init__(
self, message: Optional[object] = None, response: Optional[_HttpResponseCommonAPI] = None, **kwargs: Any
) -> None:
# Don't want to document this one yet.
error_format = kwargs.get("error_format", ODataV4Format)
self.reason: Optional[str] = None
self.status_code: Optional[int] = None
self.response: Optional[_HttpResponseCommonAPI] = response
if response:
self.reason = response.reason
self.status_code = response.status_code
# old autorest are setting "error" before calling __init__, so it might be there already
# transferring into self.model
model: Optional[Any] = kwargs.pop("model", None)
self.model: Optional[Any]
if model is not None: # autorest v5
self.model = model
else: # autorest azure-core, for KV 1.0, Storage 12.0, etc.
self.model = getattr(self, "error", None)
self.error: Optional[ODataV4Format] = self._parse_odata_body(error_format, response)
# By priority, message is:
# - odatav4 message, OR
# - parameter "message", OR
# - generic meassage using "reason"
if self.error:
message = str(self.error)
else:
message = message or "Operation returned an invalid status '{}'".format(self.reason)
super(HttpResponseError, self).__init__(message=message, **kwargs)
@staticmethod
def _parse_odata_body(
error_format: Type[ODataV4Format], response: Optional[_HttpResponseCommonAPI]
) -> Optional[ODataV4Format]:
try:
# https://github.com/python/mypy/issues/14743#issuecomment-1664725053
odata_json = json.loads(response.text()) # type: ignore
return error_format(odata_json)
except Exception: # pylint: disable=broad-except
# If the body is not JSON valid, just stop now
pass
return None
def __str__(self) -> str:
retval = super(HttpResponseError, self).__str__()
try:
# https://github.com/python/mypy/issues/14743#issuecomment-1664725053
body = self.response.text() # type: ignore
if body and not self.error:
return "{}\nContent: {}".format(retval, body)[:2048]
except Exception: # pylint: disable=broad-except
pass
return retval
[docs]class DecodeError(HttpResponseError):
"""Error raised during response deserialization."""
class IncompleteReadError(DecodeError):
"""Error raised if peer closes the connection before we have received the complete message body."""
[docs]class ResourceExistsError(HttpResponseError):
"""An error response with status code 4xx.
This will not be raised directly by the Azure core pipeline."""
[docs]class ResourceNotFoundError(HttpResponseError):
"""An error response, typically triggered by a 412 response (for update) or 404 (for get/post)"""
[docs]class ClientAuthenticationError(HttpResponseError):
"""An error response with status code 4xx.
This will not be raised directly by the Azure core pipeline."""
[docs]class ResourceModifiedError(HttpResponseError):
"""An error response with status code 4xx, typically 412 Conflict.
This will not be raised directly by the Azure core pipeline."""
[docs]class ResourceNotModifiedError(HttpResponseError):
"""An error response with status code 304.
This will not be raised directly by the Azure core pipeline."""
[docs]class TooManyRedirectsError(HttpResponseError, Generic[HTTPRequestType, HTTPResponseType]):
"""Reached the maximum number of redirect attempts.
:param history: The history of requests made while trying to fulfill the request.
:type history: list[~azure.core.pipeline.policies.RequestHistory]
"""
def __init__(
self, history: "List[RequestHistory[HTTPRequestType, HTTPResponseType]]", *args: Any, **kwargs: Any
) -> None:
self.history = history
message = "Reached maximum redirect attempts."
super(TooManyRedirectsError, self).__init__(message, *args, **kwargs)
[docs]class ODataV4Error(HttpResponseError):
"""An HTTP response error where the JSON is decoded as OData V4 error format.
http://docs.oasis-open.org/odata/odata-json-format/v4.0/os/odata-json-format-v4.0-os.html#_Toc372793091
:param ~azure.core.rest.HttpResponse response: The response object.
:ivar dict odata_json: The parsed JSON body as attribute for convenience.
:ivar str ~.code: Its value is a service-defined error code.
This code serves as a sub-status for the HTTP error code specified in the response.
:ivar str message: Human-readable, language-dependent representation of the error.
:ivar str target: The target of the particular error (for example, the name of the property in error).
This field is optional and may be None.
:ivar list[ODataV4Format] details: Array of ODataV4Format instances that MUST contain name/value pairs
for code and message, and MAY contain a name/value pair for target, as described above.
:ivar dict innererror: An object. The contents of this object are service-defined.
Usually this object contains information that will help debug the service.
"""
_ERROR_FORMAT = ODataV4Format
def __init__(self, response: _HttpResponseCommonAPI, **kwargs: Any) -> None:
# Ensure field are declared, whatever can happen afterwards
self.odata_json: Optional[Dict[str, Any]] = None
try:
self.odata_json = json.loads(response.text())
odata_message = self.odata_json.setdefault("error", {}).get("message")
except Exception: # pylint: disable=broad-except
# If the body is not JSON valid, just stop now
odata_message = None
self.code: Optional[str] = None
message: Optional[str] = kwargs.get("message", odata_message)
self.target: Optional[str] = None
self.details: Optional[List[Any]] = []
self.innererror: Optional[Mapping[str, Any]] = {}
if message and "message" not in kwargs:
kwargs["message"] = message
super(ODataV4Error, self).__init__(response=response, **kwargs)
self._error_format: Optional[Union[str, ODataV4Format]] = None
if self.odata_json:
try:
error_node = self.odata_json["error"]
self._error_format = self._ERROR_FORMAT(error_node)
self.__dict__.update({k: v for k, v in self._error_format.__dict__.items() if v is not None})
except Exception: # pylint: disable=broad-except
_LOGGER.info("Received error message was not valid OdataV4 format.")
self._error_format = "JSON was invalid for format " + str(self._ERROR_FORMAT)
def __str__(self) -> str:
if self._error_format:
return str(self._error_format)
return super(ODataV4Error, self).__str__()
[docs]class StreamConsumedError(AzureError):
"""Error thrown if you try to access the stream of a response once consumed.
It is thrown if you try to read / stream an ~azure.core.rest.HttpResponse or
~azure.core.rest.AsyncHttpResponse once the response's stream has been consumed.
:param response: The response that triggered the exception.
:type response: ~azure.core.rest.HttpResponse or ~azure.core.rest.AsyncHttpResponse
"""
def __init__(self, response: _HttpResponseCommonAPI) -> None:
message = (
"You are attempting to read or stream the content from request {}. "
"You have likely already consumed this stream, so it can not be accessed anymore.".format(response.request)
)
super(StreamConsumedError, self).__init__(message)
[docs]class StreamClosedError(AzureError):
"""Error thrown if you try to access the stream of a response once closed.
It is thrown if you try to read / stream an ~azure.core.rest.HttpResponse or
~azure.core.rest.AsyncHttpResponse once the response's stream has been closed.
:param response: The response that triggered the exception.
:type response: ~azure.core.rest.HttpResponse or ~azure.core.rest.AsyncHttpResponse
"""
def __init__(self, response: _HttpResponseCommonAPI) -> None:
message = (
"The content for response from request {} can no longer be read or streamed, since the "
"response has already been closed.".format(response.request)
)
super(StreamClosedError, self).__init__(message)
[docs]class ResponseNotReadError(AzureError):
"""Error thrown if you try to access a response's content without reading first.
It is thrown if you try to access an ~azure.core.rest.HttpResponse or
~azure.core.rest.AsyncHttpResponse's content without first reading the response's bytes in first.
:param response: The response that triggered the exception.
:type response: ~azure.core.rest.HttpResponse or ~azure.core.rest.AsyncHttpResponse
"""
def __init__(self, response: _HttpResponseCommonAPI) -> None:
message = (
"You have not read in the bytes for the response from request {}. "
"Call .read() on the response first.".format(response.request)
)
super(ResponseNotReadError, self).__init__(message)
[docs]class SerializationError(ValueError):
"""Raised if an error is encountered during serialization."""
[docs]class DeserializationError(ValueError):
"""Raised if an error is encountered during deserialization."""