[grpcio] ServerInterceptor and GenericRpcHandler should not be generic (#14785)

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

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, cast
+from typing import cast
 from typing_extensions import assert_type
 
 import grpc.aio
@@ -9,7 +9,7 @@ import grpc.aio
 client_interceptors: list[grpc.aio.ClientInterceptor] = []
 grpc.aio.insecure_channel("target", interceptors=client_interceptors)
 
-server_interceptors: list[grpc.aio.ServerInterceptor[Any, Any]] = []
+server_interceptors: list[grpc.aio.ServerInterceptor] = []
 grpc.aio.server(interceptors=server_interceptors)
 
 

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

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import cast
+from typing import Any, cast
 from typing_extensions import assert_type
 
 import grpc
@@ -19,11 +19,11 @@ def unary_unary_call(rq: Request, ctx: grpc.ServicerContext) -> Response:
     return Response()
 
 
-class ServiceHandler(grpc.ServiceRpcHandler[Request, Response]):
+class ServiceHandler(grpc.ServiceRpcHandler):
     def service_name(self) -> str:
         return "hello"
 
-    def service(self, handler_call_details: grpc.HandlerCallDetails) -> grpc.RpcMethodHandler[Request, Response] | None:
+    def service(self, handler_call_details: grpc.HandlerCallDetails) -> grpc.RpcMethodHandler[Any, Any] | None:
         rpc = grpc.RpcMethodHandler[Request, Response]()
         rpc.unary_unary = unary_unary_call
         return rpc

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

@@ -1,22 +1,35 @@
 from __future__ import annotations
 
 from collections.abc import Callable
+from concurrent.futures.thread import ThreadPoolExecutor
+from typing import Awaitable, TypeVar
 
 import grpc
+import grpc.aio
+
+RequestT = TypeVar("RequestT")
+ResponseT = TypeVar("ResponseT")
 
 
-class Request:
-    pass
-
-
-class Response:
-    pass
-
-
-class NoopInterceptor(grpc.ServerInterceptor[Request, Response]):
+class NoopInterceptor(grpc.ServerInterceptor):
     def intercept_service(
         self,
-        continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[Request, Response] | None],
+        continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[RequestT, ResponseT] | None],
         handler_call_details: grpc.HandlerCallDetails,
-    ) -> grpc.RpcMethodHandler[Request, Response] | None:
+    ) -> grpc.RpcMethodHandler[RequestT, ResponseT] | None:
         return continuation(handler_call_details)
+
+
+grpc.server(interceptors=[NoopInterceptor()], thread_pool=ThreadPoolExecutor())
+
+
+class NoopAioInterceptor(grpc.aio.ServerInterceptor):
+    async def intercept_service(
+        self,
+        continuation: Callable[[grpc.HandlerCallDetails], Awaitable[grpc.RpcMethodHandler[RequestT, ResponseT]]],
+        handler_call_details: grpc.HandlerCallDetails,
+    ) -> grpc.RpcMethodHandler[RequestT, ResponseT]:
+        return await continuation(handler_call_details)
+
+
+grpc.aio.server(interceptors=[NoopAioInterceptor()])