feat: add support for 'error_info'

google/api_core/exceptions.py

@@ -104,6 +104,8 @@ class GoogleAPICallError(GoogleAPIError, metaclass=_GoogleAPICallErrorMeta):
  details (Sequence[Any]): An optional list of objects defined in google.rpc.error_details.
  response (Union[requests.Request, grpc.Call]): The response or
  gRPC call metadata.
+ error_info (Union[error_details_pb2.ErrorInfo, None]): An optional object containing error info
+ (google.rpc.error_details.ErrorInfo).
  """
 
  code: Union[int, None] = None
@@ -122,12 +124,13 @@ class GoogleAPICallError(GoogleAPIError, metaclass=_GoogleAPICallErrorMeta):
  This may be ``None`` if the exception does not match up to a gRPC error.
  """
 
- def __init__(self, message, errors=(), details=(), response=None):
+ def __init__(self, message, errors=(), details=(), response=None, error_info=None):
  super(GoogleAPICallError, self).__init__(message)
  self.message = message
  """str: The exception message."""
  self._errors = errors
  self._details = details
+ self._error_info = error_info
  self._response = response
 
  def __str__(self):
@@ -145,6 +148,48 @@ def errors(self):
  """
  return list(self._errors)
 
+ @property
+ def reason(self):
+ """The reason of the error.
+
+ Reference:
+ https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto#L112
+
+ Returns:
+ Union[str, None]: An optional string containing reason of the error.
+ """
+ if not self._error_info:
+ return None
+ return self._error_info.reason
+
+ @property
+ def domain(self):
+ """The logical grouping to which the "reason" belongs.
+
+ Reference:
+ https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto#L112
+
+ Returns:
+ Union[str, None]: An optional string containing a logical grouping to which the "reason" belongs.
+ """
+ if not self._error_info:
+ return None
+ return self._error_info.domain
+
+ @property
+ def metadata(self):
+ """Additional structured details about this error
+
+ Reference:
+ https://github.com/googleapis/googleapis/blob/master/google/rpc/error_details.proto#L112
+
+ Returns:
+ Union[Dict[str, str], None]: An optional object containing structured details about the error.
+ """
+ if not self._error_info:
+ return None
+ return self._error_info.metadata
+
  @property
  def details(self):
  """Information contained in google.rpc.status.details.
@@ -433,13 +478,25 @@ def from_http_response(response):
  errors = payload.get("error", {}).get("errors", ())
  # In JSON, details are already formatted in developer-friendly way.
  details = payload.get("error", {}).get("details", ())
-
+ error_info = list(
+ filter(
+ lambda detail: detail.get("@type", "")
+ == "type.googleapis.com/google.rpc.ErrorInfo",
+ details,
+ )
+ )
+ error_info = error_info[0] if error_info else None
  message = "{method} {url}: {error}".format(
  method=response.request.method, url=response.request.url, error=error_message
  )
 
  exception = from_http_status(
- response.status_code, message, errors=errors, details=details, response=response
+ response.status_code,
+ message,
+ errors=errors,
+ details=details,
+ response=response,
+ error_info=error_info,
  )
  return exception
 
@@ -490,10 +547,10 @@ def _parse_grpc_error_details(rpc_exc):
  try:
  status = rpc_status.from_call(rpc_exc)
  except NotImplementedError: # workaround
- return []
+ return [], None
 
  if not status:
- return []
+ return [], None
 
  possible_errors = [
  error_details_pb2.BadRequest,
@@ -507,6 +564,7 @@ def _parse_grpc_error_details(rpc_exc):
  error_details_pb2.Help,
  error_details_pb2.LocalizedMessage,
  ]
+ error_info = None
  error_details = []
  for detail in status.details:
  matched_detail_cls = list(
@@ -519,7 +577,9 @@ def _parse_grpc_error_details(rpc_exc):
  info = matched_detail_cls[0]()
  detail.Unpack(info)
  error_details.append(info)
- return error_details
+ if isinstance(info, error_details_pb2.ErrorInfo):
+ error_info = info
+ return error_details, error_info
 
 
 def from_grpc_error(rpc_exc):
@@ -535,12 +595,14 @@ def from_grpc_error(rpc_exc):
  # NOTE(lidiz) All gRPC error shares the parent class grpc.RpcError.
  # However, check for grpc.RpcError breaks backward compatibility.
  if isinstance(rpc_exc, grpc.Call) or _is_informative_grpc_error(rpc_exc):
+ details, err_info = _parse_grpc_error_details(rpc_exc)
  return from_grpc_status(
  rpc_exc.code(),
  rpc_exc.details(),
  errors=(rpc_exc,),
- details=_parse_grpc_error_details(rpc_exc),
+ details=details,
  response=rpc_exc,
+ error_info=err_info,
  )
  else:
  return GoogleAPICallError(str(rpc_exc), errors=(rpc_exc,), response=rpc_exc)

tests/unit/test_exceptions.py

@@ -275,31 +275,56 @@ def create_bad_request_details():
  return status_detail
 
 
+def create_error_info_details():
+ info = error_details_pb2.ErrorInfo(
+ reason="SERVICE_DISABLED",
+ domain="googleapis.com",
+ metadata={
+ "consumer": "projects/455411330361",
+ "service": "translate.googleapis.com",
+ },
+ )
+ status_detail = any_pb2.Any()
+ status_detail.Pack(info)
+ return status_detail
+
+
 def test_error_details_from_rest_response():
  bad_request_detail = create_bad_request_details()
+ error_info_detail = create_error_info_details()
  status = status_pb2.Status()
  status.code = 3
  status.message = (
  "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set."
  )
  status.details.append(bad_request_detail)
+ status.details.append(error_info_detail)
 
  # See JSON schema in https://cloud.google.com/apis/design/errors#http_mapping
  http_response = make_response(
- json.dumps({"error": json.loads(json_format.MessageToJson(status))}).encode(
-"utf-8"
- )
+ json.dumps(
+{"error": json.loads(json_format.MessageToJson(status, sort_keys=True))}
+ ).encode("utf-8")
  )
  exception = exceptions.from_http_response(http_response)
- want_error_details = [json.loads(json_format.MessageToJson(bad_request_detail))]
+ want_error_details = [
+ json.loads(json_format.MessageToJson(bad_request_detail)),
+ json.loads(json_format.MessageToJson(error_info_detail)),
+ ]
  assert want_error_details == exception.details
+
  # 404 POST comes from make_response.
  assert str(exception) == (
  "404 POST https://example.com/: 3 INVALID_ARGUMENT:"
  " One of content, or gcs_content_uri must be set."
  " [{'@type': 'type.googleapis.com/google.rpc.BadRequest',"
- " 'fieldViolations': [{'field': 'document.content',"
- " 'description': 'Must have some text content to annotate.'}]}]"
+ " 'fieldViolations': [{'description': 'Must have some text content to annotate.',"
+ " 'field': 'document.content'}]},"
+ " {'@type': 'type.googleapis.com/google.rpc.ErrorInfo',"
+ " 'domain': 'googleapis.com',"
+ " 'metadata': {'consumer': 'projects/455411330361',"
+ " 'service': 'translate.googleapis.com'},"
+ " 'reason': 'SERVICE_DISABLED'}]"
  )
 
 
@@ -311,6 +336,11 @@ def test_error_details_from_v1_rest_response():
  )
  exception = exceptions.from_http_response(response)
  assert exception.details == []
+ assert (
+ exception.reason is None
+ and exception.domain is None
+ and exception.metadata is None
+ )
 
 
 @pytest.mark.skipif(grpc is None, reason="gRPC not importable")
@@ -320,9 +350,10 @@ def test_error_details_from_grpc_response():
  status.message = (
  "3 INVALID_ARGUMENT: One of content, or gcs_content_uri must be set."
  )
- status_detail = create_bad_request_details()
- status.details.append(status_detail)
-
+ status_br_detail = create_bad_request_details()
+ status_ei_detail = create_error_info_details()
+ status.details.append(status_br_detail)
+ status.details.append(status_ei_detail)
  # Actualy error doesn't matter as long as its grpc.Call,
  # because from_call is mocked.
  error = mock.create_autospec(grpc.Call, instance=True)
@@ -331,8 +362,13 @@ def test_error_details_from_grpc_response():
  exception = exceptions.from_grpc_error(error)
 
  bad_request_detail = error_details_pb2.BadRequest()
- status_detail.Unpack(bad_request_detail)
- assert exception.details == [bad_request_detail]
+ error_info_detail = error_details_pb2.ErrorInfo()
+ status_br_detail.Unpack(bad_request_detail)
+ status_ei_detail.Unpack(error_info_detail)
+ assert exception.details == [bad_request_detail, error_info_detail]
+ assert exception.reason == error_info_detail.reason
+ assert exception.domain == error_info_detail.domain
+ assert exception.metadata == error_info_detail.metadata
 
 
 @pytest.mark.skipif(grpc is None, reason="gRPC not importable")
@@ -351,3 +387,8 @@ def test_error_details_from_grpc_response_unknown_error():
  m.return_value = status
  exception = exceptions.from_grpc_error(error)
  assert exception.details == [status_detail]
+ assert (
+ exception.reason is None
+ and exception.domain is None
+ and exception.metadata is None
+ )