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,20 +124,57 @@ 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._response = response
+        self._error_info = error_info
 
     def __str__(self):
         if self.details:
             return "{} {} {}".format(self.code, self.message, self.details)
         else:
             return "{} {}".format(self.code, self.message)
 
+    @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.
+        """
+        return self._error_info.reason if self._error_info else None
+
+    @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.
+        """
+        return self._error_info.domain if self._error_info else None
+
+    @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.
+        """
+        return self._error_info.metadata if self._error_info else None
+
     @property
     def errors(self):
         """Detailed error information.
@@ -433,13 +472,26 @@ 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
+        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 +542,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 +559,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 +572,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 +590,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,8 +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.
@@ -331,8 +363,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 +388,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
+    )