Create exceptions (and bases) for RPC-only errors in core

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

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):