Import gRPC stubs from the grpc-stubs project

pyrightconfig.stricter.json

@@ -51,6 +51,7 @@
         "stubs/gdb",
         "stubs/geopandas",
         "stubs/google-cloud-ndb",
+        "stubs/grpcio/grpc/__init__.pyi",
         "stubs/hdbcli/hdbcli/dbapi.pyi",
         "stubs/html5lib",
         "stubs/httplib2",

stubs/grpcio/@tests/stubtest_allowlist.txt

@@ -0,0 +1,11 @@
+# Error: is not present at runtime
+# =============================
+# Error class attributes that aren't defined.
+grpc.RpcError.code
+grpc.RpcError.details
+grpc.RpcError.trailing_metadata
+
+# Error: is inconsistent
+# =============================
+# Stub class is incomplete.
+grpc_reflection.v1alpha._base.BaseReflectionServicer.__init__

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

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from typing import Any, cast
+from typing_extensions import assert_type
+
+import grpc.aio
+
+# Interceptor casts
+client_interceptors: list[grpc.aio.ClientInterceptor] = []
+grpc.aio.insecure_channel("target", interceptors=client_interceptors)
+
+server_interceptors: list[grpc.aio.ServerInterceptor[Any, Any]] = []
+grpc.aio.server(interceptors=server_interceptors)
+
+
+# Metadata
+async def metadata() -> None:
+    metadata = await cast(grpc.aio.Call, None).initial_metadata()
+    assert_type(metadata["foo"], grpc.aio._MetadataValue)
+    for k in metadata:
+        assert_type(k, str)
+
+    for k, v in metadata.items():
+        assert_type(k, str)
+        assert_type(v, grpc.aio._MetadataValue)

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

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from typing import Protocol, cast
+from typing_extensions import assert_type
+
+import grpc.aio
+
+
+class DummyRequest:
+    pass
+
+
+class DummyReply:
+    pass
+
+
+class DummyServiceStub(Protocol):
+    UnaryUnary: grpc.aio.UnaryUnaryMultiCallable[DummyRequest, DummyReply]
+    UnaryStream: grpc.aio.UnaryStreamMultiCallable[DummyRequest, DummyReply]
+    StreamUnary: grpc.aio.StreamUnaryMultiCallable[DummyRequest, DummyReply]
+    StreamStream: grpc.aio.StreamStreamMultiCallable[DummyRequest, DummyReply]
+
+
+stub = cast(DummyServiceStub, None)
+req = DummyRequest()
+
+
+async def async_context() -> None:
+    assert_type(await stub.UnaryUnary(req), DummyReply)
+
+    async for resp in stub.UnaryStream(req):
+        assert_type(resp, DummyReply)
+
+    assert_type(await stub.StreamUnary(iter([req])), DummyReply)
+
+    async for resp in stub.StreamStream(iter([req])):
+        assert_type(resp, DummyReply)

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

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import Optional, cast
+from typing_extensions import assert_type
+
+import grpc
+
+# Channel options:
+assert_type(grpc.insecure_channel("target", ()), grpc.Channel)
+assert_type(grpc.insecure_channel("target", (("a", "b"),)), grpc.Channel)
+assert_type(grpc.insecure_channel("target", (("a", "b"), ("c", "d"))), grpc.Channel)
+
+# Local channel credentials:
+creds = grpc.local_channel_credentials(grpc.LocalConnectionType.LOCAL_TCP)
+assert_type(creds, grpc.ChannelCredentials)
+
+# Other credential types:
+assert_type(grpc.alts_channel_credentials(), grpc.ChannelCredentials)
+assert_type(grpc.alts_server_credentials(), grpc.ServerCredentials)
+assert_type(grpc.compute_engine_channel_credentials(grpc.CallCredentials("")), grpc.ChannelCredentials)
+assert_type(grpc.insecure_server_credentials(), grpc.ServerCredentials)
+
+# XDS credentials:
+assert_type(
+    grpc.xds_channel_credentials(grpc.local_channel_credentials(grpc.LocalConnectionType.LOCAL_TCP)), grpc.ChannelCredentials
+)
+assert_type(grpc.xds_server_credentials(grpc.insecure_server_credentials()), grpc.ServerCredentials)
+
+# Channel ready future
+channel = grpc.insecure_channel("target", ())
+assert_type(grpc.channel_ready_future(channel).result(), None)
+
+# Channel options supports list:
+assert_type(grpc.insecure_channel("target", []), grpc.Channel)
+assert_type(grpc.insecure_channel("target", [("a", "b")]), grpc.Channel)
+assert_type(grpc.insecure_channel("target", [("a", "b"), ("c", "d")]), grpc.Channel)
+
+# Client call details optionals:
+call_details = grpc.ClientCallDetails()
+assert_type(call_details.method, str)
+assert_type(call_details.timeout, Optional[float])
+
+# Call iterator
+call_iter = cast(grpc._CallIterator[str], None)
+for call in call_iter:
+    assert_type(call, str)

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

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+from typing import cast
+from typing_extensions import assert_type
+
+import grpc
+
+
+class Request:
+    pass
+
+
+class Response:
+    pass
+
+
+def unary_unary_call(rq: Request, ctx: grpc.ServicerContext) -> Response:
+    assert_type(rq, Request)
+    return Response()
+
+
+class ServiceHandler(grpc.ServiceRpcHandler[Request, Response]):
+    def service_name(self) -> str:
+        return "hello"
+
+    def service(self, handler_call_details: grpc.HandlerCallDetails) -> grpc.RpcMethodHandler[Request, Response] | None:
+        rpc = grpc.RpcMethodHandler[Request, Response]()
+        rpc.unary_unary = unary_unary_call
+        return rpc
+
+
+h = ServiceHandler()
+ctx = cast(grpc.ServicerContext, None)
+svc = h.service(grpc.HandlerCallDetails())
+if svc is not None and svc.unary_unary is not None:
+    svc.unary_unary(Request(), ctx)

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

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import Protocol, cast
+from typing_extensions import assert_type
+
+import grpc
+
+
+class DummyRequest:
+    pass
+
+
+class DummyReply:
+    pass
+
+
+class DummyServiceStub(Protocol):
+    UnaryUnary: grpc.UnaryUnaryMultiCallable[DummyRequest, DummyReply]
+    UnaryStream: grpc.UnaryStreamMultiCallable[DummyRequest, DummyReply]
+    StreamUnary: grpc.StreamUnaryMultiCallable[DummyRequest, DummyReply]
+    StreamStream: grpc.StreamStreamMultiCallable[DummyRequest, DummyReply]
+
+
+stub = cast(DummyServiceStub, None)
+req = DummyRequest()
+
+assert_type(stub.UnaryUnary(req), DummyReply)
+
+for resp in stub.UnaryStream(req):
+    assert_type(resp, DummyReply)
+
+assert_type(stub.StreamUnary(iter([req])), DummyReply)
+
+for resp in stub.StreamStream(iter([req])):
+    assert_type(resp, DummyReply)

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

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import cast
+
+import grpc
+from grpc_reflection.v1alpha.reflection import enable_server_reflection
+
+server = cast(grpc.Server, None)
+enable_server_reflection(["foo"], server, None)

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

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from typing import cast
+
+import grpc.aio
+from grpc_reflection.v1alpha.reflection import enable_server_reflection
+
+server = cast(grpc.aio.Server, None)
+enable_server_reflection(["foo"], server, None)

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

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Any
+
+import grpc
+
+
+@grpc.Call.register
+class CallProxy:
+    def __init__(self, target: grpc.Call) -> None:
+        self._target = target
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._target, name)

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

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import grpc
+
+
+class Request:
+    pass
+
+
+class Response:
+    pass
+
+
+class NoopInterceptor(grpc.ServerInterceptor[Request, Response]):
+    def intercept_service(
+        self,
+        continuation: Callable[[grpc.HandlerCallDetails], grpc.RpcMethodHandler[Request, Response] | None],
+        handler_call_details: grpc.HandlerCallDetails,
+    ) -> grpc.RpcMethodHandler[Request, Response] | None:
+        return continuation(handler_call_details)

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

@@ -0,0 +1,8 @@
+from __future__ import annotations
+
+from grpc import Status
+from grpc_status import to_status
+
+# XXX: to_status actually expects a "google.rpc.status.Status",
+# but the stubs for that aren't present yet.
+status: Status = to_status(None)

stubs/grpcio/METADATA.toml

@@ -0,0 +1,15 @@
+version = "1.*"
+upstream_repository = "https://github.com/grpc/grpc"
+partial_stub = true
+requires = [
+    "types-protobuf",
+]
+
+[tool.stubtest]
+ignore_missing_stub = true
+stubtest_requirements = [
+    "grpcio-channelz",
+    "grpcio-health-checking",
+    "grpcio-reflection",
+    "grpcio-status",
+]