[AbortError] Reapply "[AbortError] And and check AbortError while abo…

…rt" (grpc#34525)

Reverts grpc#34515

This PR reapplies AbortError change as the previous one was reverted.

This change was mentioned in this gRFC: [L105: Python Add New Error Types](https://github.com/grpc/proposal/blob/master/L105-python-expose-new-error-types.md)

Closes grpc#34525

COPYBARA_INTEGRATE_REVIEW=grpc#34525 from grpc:revert-34515-revert-33969-checkAbortError ce63ab1
PiperOrigin-RevId: 613718295

src/python/grpcio/grpc/BUILD.bazel

@@ -99,6 +99,11 @@ py_library(
     srcs = ["_observability.py"],
 )
 
+py_library(
+    name = "errors",
+    srcs = ["_errors.py"],
+)
+
 py_library(
     name = "grpcio",
     srcs = ["__init__.py"],
@@ -115,6 +120,7 @@ py_library(
         ":auth",
         ":channel",
         ":compression",
+        ":errors",
         ":interceptor",
         ":plugin_wrapping",
         ":server",

src/python/grpcio/grpc/__init__.py

@@ -21,6 +21,9 @@
 
 from grpc import _compression
 from grpc._cython import cygrpc as _cygrpc
+from grpc._errors import AbortError
+from grpc._errors import BaseError
+from grpc._errors import RpcError
 from grpc._runtime_protos import protos
 from grpc._runtime_protos import protos_and_services
 from grpc._runtime_protos import services
@@ -307,13 +310,6 @@ class Status(abc.ABC):
     """
 
 
-#############################  gRPC Exceptions  ################################
-
-
-class RpcError(Exception):
-    """Raised by the gRPC library to indicate non-OK-status RPC termination."""
-
-
 ##############################  Shared Context  ################################
 
 
@@ -1241,8 +1237,8 @@ def abort(self, code, details):
             termination of the RPC.
 
         Raises:
-          Exception: An exception is always raised to signal the abortion the
-            RPC to the gRPC runtime.
+          AbortError: A grpc.AbortError is always raised to signal the abortion
+            the RPC to the gRPC runtime.
         """
         raise NotImplementedError()
 
@@ -1260,8 +1256,8 @@ def abort_with_status(self, status):
             StatusCode.OK.
 
         Raises:
-          Exception: An exception is always raised to signal the abortion the
-            RPC to the gRPC runtime.
+          AbortError: A grpc.AbortError is always raised to signal the abortion
+            the RPC to the gRPC runtime.
         """
         raise NotImplementedError()
 
@@ -2273,6 +2269,8 @@ class Compression(enum.IntEnum):
     "ServiceRpcHandler",
     "Server",
     "ServerInterceptor",
+    "AbortError",
+    "BaseError",
     "unary_unary_rpc_method_handler",
     "unary_stream_rpc_method_handler",
     "stream_unary_rpc_method_handler",

src/python/grpcio/grpc/_channel.py

@@ -369,7 +369,9 @@ def _rpc_state_string(class_name: str, rpc_state: _RPCState) -> str:
             )
 
 
-class _InactiveRpcError(grpc.RpcError, grpc.Call, grpc.Future):
+class _InactiveRpcError(
+    grpc.RpcError, grpc.Call, grpc.Future
+):  # pylint: disable=too-many-ancestors
     """An RPC error not tied to the execution of a particular RPC.
 
     The RPC represented by the state object must not be in-progress or

src/python/grpcio/grpc/_cython/_cygrpc/aio/call.pyx.pxi

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from grpc._errors import InternalError
 
 _EMPTY_FLAGS = 0
 _EMPTY_MASK = 0

src/python/grpcio/grpc/_cython/_cygrpc/aio/callback_common.pyx.pxi

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from grpc._errors import InternalError
 
 cdef class CallbackFailureHandler:
 

src/python/grpcio/grpc/_cython/_cygrpc/aio/channel.pyx.pxi

@@ -13,6 +13,7 @@
 # limitations under the License.
 #
 
+from grpc._errors import UsageError
 
 class _WatchConnectivityFailed(Exception):
     """Dedicated exception class for watch connectivity failed.

src/python/grpcio/grpc/_cython/_cygrpc/aio/common.pyx.pxi

@@ -82,30 +82,6 @@ _COMPRESSION_METADATA_STRING_MAPPING = {
     CompressionAlgorithm.gzip: 'gzip',
 }
 
-class BaseError(Exception):
-    """The base class for exceptions generated by gRPC AsyncIO stack."""
-
-
-class UsageError(BaseError):
-    """Raised when the usage of API by applications is inappropriate.
-
-    For example, trying to invoke RPC on a closed channel, mixing two styles
-    of streaming API on the client side. This exception should not be
-    suppressed.
-    """
-
-
-class AbortError(BaseError):
-    """Raised when calling abort in servicer methods.
-
-    This exception should not be suppressed. Applications may catch it to
-    perform certain clean-up logic, and then re-raise it.
-    """
-
-
-class InternalError(BaseError):
-    """Raised upon unexpected errors in native code."""
-
 
 def schedule_coro_threadsafe(object coro, object loop):
     try:

src/python/grpcio/grpc/_cython/_cygrpc/aio/server.pyx.pxi

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 
+from grpc._errors import BaseError, AbortError, InternalError, UsageError
+
 import inspect
 import traceback
 import functools

src/python/grpcio/grpc/_cython/_cygrpc/server.pyx.pxi

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from grpc._errors import InternalError, UsageError
 
 cdef class Server:
 

src/python/grpcio/grpc/_errors.py

@@ -0,0 +1,57 @@
+# Copyright 2024 The gRPC authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+#############################  gRPC Exceptions  ################################
+
+
+class BaseError(Exception):
+    """
+    The base class for exceptions generated by gRPC.
+    """
+
+
+class UsageError(BaseError):
+    """
+    Raised when the usage of API by applications is inappropriate.
+    For example, trying to invoke RPC on a closed channel, mixing two styles
+    of streaming API on the client side. This exception should not be
+    suppressed.
+    """
+
+
+class AbortError(BaseError):
+    """
+    Raised when calling abort in servicer methods.
+    This exception should not be suppressed. Applications may catch it to
+    perform certain clean-up logic, and then re-raise it.
+    """
+
+
+class InternalError(BaseError):
+    """
+    Raised upon unexpected errors in native code.
+    """
+
+
+class RpcError(BaseError):
+    """Raised by the gRPC library to indicate non-OK-status RPC termination."""
+
+
+__all__ = (
+    "BaseError",
+    "UsageError",
+    "AbortError",
+    "InternalError",
+    "RpcError",
+)

src/python/grpcio/grpc/_server.py

@@ -42,6 +42,7 @@
 from grpc import _compression  # pytype: disable=pyi-error
 from grpc import _interceptor  # pytype: disable=pyi-error
 from grpc._cython import cygrpc
+from grpc._errors import AbortError
 from grpc._typing import ArityAgnosticMethodHandler
 from grpc._typing import ChannelArgumentType
 from grpc._typing import DeserializingFunction
@@ -404,7 +405,7 @@ def abort(self, code: grpc.StatusCode, details: str) -> None:
             self._state.code = code
             self._state.details = _common.encode(details)
             self._state.aborted = True
-            raise Exception()
+            raise AbortError()
 
     def abort_with_status(self, status: grpc.Status) -> None:
         self._state.trailing_metadata = status.trailing_metadata
@@ -557,6 +558,15 @@ def _call_behavior(
         except Exception as exception:  # pylint: disable=broad-except
             with state.condition:
                 if state.aborted:
+                    if not isinstance(exception, AbortError):
+                        try:
+                            details = f"Exception happened while aborting: {exception}"
+                        except Exception:  # pylint: disable=broad-except
+                            details = (
+                                "Calling abort raised unprintable Exception!"
+                            )
+                            traceback.print_exc()
+                        _LOGGER.exception(details)
                     _abort(
                         state,
                         rpc_event.call,

src/python/grpcio/grpc/aio/__init__.py

@@ -20,13 +20,13 @@
 from typing import Any, Optional, Sequence, Tuple
 
 import grpc
-from grpc._cython.cygrpc import AbortError
-from grpc._cython.cygrpc import BaseError
 from grpc._cython.cygrpc import EOF
-from grpc._cython.cygrpc import InternalError
-from grpc._cython.cygrpc import UsageError
 from grpc._cython.cygrpc import init_grpc_aio
 from grpc._cython.cygrpc import shutdown_grpc_aio
+from grpc._errors import AbortError
+from grpc._errors import BaseError
+from grpc._errors import InternalError
+from grpc._errors import UsageError
 
 from ._base_call import Call
 from ._base_call import RpcContext

src/python/grpcio/grpc/aio/_call.py

@@ -24,6 +24,8 @@
 import grpc
 from grpc import _common
 from grpc._cython import cygrpc
+from grpc._errors import InternalError
+from grpc._errors import UsageError
 
 from . import _base_call
 from ._metadata import Metadata
@@ -337,7 +339,7 @@ def _update_response_style(self, style: _APIStyle):
         if self._response_style is _APIStyle.UNKNOWN:
             self._response_style = style
         elif self._response_style is not style:
-            raise cygrpc.UsageError(_API_STYLE_ERROR)
+            raise UsageError(_API_STYLE_ERROR)
 
     def cancel(self) -> bool:
         if super().cancel():
@@ -418,7 +420,7 @@ def _init_stream_request_mixin(
 
     def _raise_for_different_style(self, style: _APIStyle):
         if self._request_style is not style:
-            raise cygrpc.UsageError(_API_STYLE_ERROR)
+            raise UsageError(_API_STYLE_ERROR)
 
     def cancel(self) -> bool:
         if super().cancel():
@@ -490,7 +492,7 @@ async def _write(self, request: RequestType) -> None:
         )
         try:
             await self._cython_call.send_serialized_message(serialized_request)
-        except cygrpc.InternalError as err:
+        except InternalError as err:
             self._cython_call.set_internal_error(str(err))
             await self._raise_for_status()
         except asyncio.CancelledError:

src/python/grpcio/grpc/aio/_channel.py

@@ -22,6 +22,7 @@
 from grpc import _compression
 from grpc import _grpcio_metadata
 from grpc._cython import cygrpc
+from grpc._errors import InternalError
 
 from . import _base_call
 from . import _base_channel
@@ -431,7 +432,7 @@ async def _close(self, grace):  # pylint: disable=too-many-branches
                             continue
                     else:
                         # Unidentified Call object
-                        raise cygrpc.InternalError(
+                        raise InternalError(
                             f"Unrecognized call object: {candidate}"
                         )
 

src/python/grpcio/grpc/aio/_interceptor.py

@@ -30,6 +30,7 @@
 
 import grpc
 from grpc._cython import cygrpc
+from grpc._errors import UsageError
 
 from . import _base_call
 from ._call import AioRpcError
@@ -562,7 +563,7 @@ async def write(self, request: RequestType) -> None:
         # should be expected through an iterators provided
         # by the caller.
         if self._write_to_iterator_queue is None:
-            raise cygrpc.UsageError(_API_STYLE_ERROR)
+            raise UsageError(_API_STYLE_ERROR)
 
         try:
             call = await self._interceptors_task
@@ -588,7 +589,7 @@ async def done_writing(self) -> None:
         # should be expected through an iterators provided
         # by the caller.
         if self._write_to_iterator_queue is None:
-            raise cygrpc.UsageError(_API_STYLE_ERROR)
+            raise UsageError(_API_STYLE_ERROR)
 
         try:
             call = await self._interceptors_task