[grpcio] Type aio ServerInterceptor with an async method handler

Author: wolfgang-aura

stubs/grpcio/@tests/test_cases/check_server_interceptor.py

@@ -6,6 +6,7 @@
 
 import grpc
 import grpc.aio
+from grpc.aio import _AsyncRpcMethodHandler
 
 RequestT = TypeVar("RequestT")
 ResponseT = TypeVar("ResponseT")
@@ -26,9 +27,9 @@ def intercept_service(
 class NoopAioInterceptor(grpc.aio.ServerInterceptor):
     async def intercept_service(
         self,
-        continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler[RequestT, ResponseT] | None]],
+        continuation: Callable[[grpc.HandlerCallDetails], Awaitable[_AsyncRpcMethodHandler[RequestT, ResponseT] | None]],
         handler_call_details: grpc.HandlerCallDetails,
-    ) -> grpc.RpcMethodHandler[RequestT, ResponseT] | None:
+    ) -> _AsyncRpcMethodHandler[RequestT, ResponseT] | None:
         return await continuation(handler_call_details)
 
 

stubs/grpcio/grpc/aio/__init__.pyi

@@ -26,7 +26,6 @@ from grpc import (
     GenericRpcHandler,
     HandlerCallDetails,
     RpcError,
-    RpcMethodHandler,
     ServerCredentials,
     Status,
     StatusCode,
@@ -437,6 +436,19 @@ class StreamStreamClientInterceptor(ClientInterceptor, metaclass=abc.ABCMeta):
 
 # Server-Side Interceptor:
 
+# Async version of grpc.RpcMethodHandler. As opposed to the former, this type
+# does not exist at runtime.
+@type_check_only
+class _AsyncRpcMethodHandler(Generic[_TRequest, _TResponse]):
+    request_streaming: bool
+    response_streaming: bool
+    request_deserializer: _Deserializer[_TRequest] | None
+    response_serializer: _Serializer[_TResponse] | None
+    unary_unary: Callable[[_TRequest, ServicerContext[_TRequest, _TResponse]], Awaitable[_TResponse]] | None
+    unary_stream: Callable[[_TRequest, ServicerContext[_TRequest, _TResponse]], AsyncIterator[_TResponse]] | None
+    stream_unary: Callable[[AsyncIterator[_TRequest], ServicerContext[_TRequest, _TResponse]], Awaitable[_TResponse]] | None
+    stream_stream: Callable[[AsyncIterator[_TRequest], ServicerContext[_TRequest, _TResponse]], AsyncIterator[_TResponse]] | None
+
 class ServerInterceptor(metaclass=abc.ABCMeta):
     # This method (not the class) is generic over _TRequest and _TResponse
     # and the types must satisfy the no-op implementation of
@@ -447,9 +459,9 @@ class ServerInterceptor(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     async def intercept_service(
         self,
-        continuation: Callable[[HandlerCallDetails], Awaitable[RpcMethodHandler[_TRequest, _TResponse] | None]],
+        continuation: Callable[[HandlerCallDetails], Awaitable[_AsyncRpcMethodHandler[_TRequest, _TResponse] | None]],
         handler_call_details: HandlerCallDetails,
-    ) -> RpcMethodHandler[_TRequest, _TResponse] | None: ...
+    ) -> _AsyncRpcMethodHandler[_TRequest, _TResponse] | None: ...
 
 # Multi-Callable Interfaces: