diff --git a/core/google/cloud/exceptions.py b/core/google/cloud/exceptions.py index 01bec56d5b27..248b54f00b86 100644 --- a/core/google/cloud/exceptions.py +++ b/core/google/cloud/exceptions.py @@ -15,6 +15,7 @@ """Custom exceptions for :mod:`google.cloud` package. See: https://cloud.google.com/storage/docs/json_api/v1/status-codes +See: https://github.com/googleapis/googleapis/blob/master/google/rpc/code.proto """ # Avoid the grpc and google.cloud.grpc collision. @@ -72,6 +73,11 @@ def errors(self): return [copy.deepcopy(error) for error in self._errors] +class GoogleCloudRPCError(GoogleCloudError): + """Base error class for RPC errors that do not map directly to HTTP errors. + """ + + class Redirection(GoogleCloudError): """Base for 3xx responses @@ -181,10 +187,24 @@ class ServiceUnavailable(ServerError): class GatewayTimeout(ServerError): - """Excepption mapping a `504 Gateway Timeout'` response.""" + """Exception mapping a '504 Gateway Timeout' response.""" code = 504 +class RPCServerError(GoogleCloudRPCError): + """Base for 5xx-like RPC errors: (abstract)""" + + +class Unknown(RPCServerError): + """Exception mapping a '2 UNKNOWN' RPC error.""" + code = 2 + + +class DataLoss(RPCServerError): + """Exception mapping a '15 DATA LOSS' RPC error.""" + code = 15 + + def make_exception(response, content, error_info=None, use_json=True): """Factory: create exception based on HTTP response code. @@ -245,8 +265,11 @@ def _walk_subclasses(klass): yield subsub -# Build the code->exception class mapping. +# Build the HTTP code->exception class mapping, excluding RPC-only exceptions. for _eklass in _walk_subclasses(GoogleCloudError): + if issubclass(_eklass, GoogleCloudRPCError): + continue + code = getattr(_eklass, 'code', None) if code is not None: _HTTP_CODE_TO_EXCEPTION[code] = _eklass diff --git a/core/unit_tests/test_exceptions.py b/core/unit_tests/test_exceptions.py index 36cdc3f3e360..79bc9d0fa491 100644 --- a/core/unit_tests/test_exceptions.py +++ b/core/unit_tests/test_exceptions.py @@ -46,6 +46,14 @@ def test_ctor_explicit(self): self.assertEqual(e.message, 'Testing') self.assertEqual(list(e.errors), [ERROR]) + def test_rpc_only_errors_not_exported(self): + from google.cloud.exceptions import GoogleCloudRPCError + from google.cloud.exceptions import _HTTP_CODE_TO_EXCEPTION + + http_exceptions = _HTTP_CODE_TO_EXCEPTION.values() + self.assertFalse( + any([issubclass(e, GoogleCloudRPCError) for e in http_exceptions])) + class Test_make_exception(unittest.TestCase):