Add partial aio typing stubs

[RobinMcCorkell]

Description of change

Added typing stubs for:

• Multi-callable
• Server
• Channel

This should be sufficient for basic mypy-protobuf usage.

Also added a test for multi-callable usage, since that's the gnarliest part.

Minimum Reproducible Example

Pull requests will not be accepted without minimum reproducible examples. "Reproducible" in this case
means the following is provided, in separate <details> blocks, using as few files as possible. Gists
and links to other repositories are not acceptable as an MRE. Pull requests without an MRE will be
immediately closed.

• One or more python files containing reproducing code.
• Full set of shell commands (POSIX shell or bash only) required to create a venv, install
  dependencies, and generate proto.

This is necessary due to the large number of PRs that have been missing tests or examples, which
causes knock-on effects for all users.

dummy_pb2_grpc.py

# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
"""Client and server classes corresponding to protobuf-defined services."""
import grpc

import dummy_pb2 as testproto_dot_grpc_dot_dummy__pb2


class DummyServiceStub(object):
    """DummyService
    """

    def __init__(self, channel):
        """Constructor.

        Args:
            channel: A grpc.Channel.
        """
        self.UnaryUnary = channel.unary_unary(
                '/dummy.DummyService/UnaryUnary',
                request_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
                response_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
                )
        self.UnaryStream = channel.unary_stream(
                '/dummy.DummyService/UnaryStream',
                request_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
                response_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
                )
        self.StreamUnary = channel.stream_unary(
                '/dummy.DummyService/StreamUnary',
                request_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
                response_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
                )
        self.StreamStream = channel.stream_stream(
                '/dummy.DummyService/StreamStream',
                request_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
                response_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
                )


class DummyServiceServicer(object):
    """DummyService
    """

    def UnaryUnary(self, request, context):
        """UnaryUnary
        """
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def UnaryStream(self, request, context):
        """UnaryStream
        """
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def StreamUnary(self, request_iterator, context):
        """StreamUnary
        """
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')

    def StreamStream(self, request_iterator, context):
        """StreamStream
        """
        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
        context.set_details('Method not implemented!')
        raise NotImplementedError('Method not implemented!')


def add_DummyServiceServicer_to_server(servicer, server):
    rpc_method_handlers = {
            'UnaryUnary': grpc.unary_unary_rpc_method_handler(
                    servicer.UnaryUnary,
                    request_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.FromString,
                    response_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.SerializeToString,
            ),
            'UnaryStream': grpc.unary_stream_rpc_method_handler(
                    servicer.UnaryStream,
                    request_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.FromString,
                    response_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.SerializeToString,
            ),
            'StreamUnary': grpc.stream_unary_rpc_method_handler(
                    servicer.StreamUnary,
                    request_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.FromString,
                    response_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.SerializeToString,
            ),
            'StreamStream': grpc.stream_stream_rpc_method_handler(
                    servicer.StreamStream,
                    request_deserializer=testproto_dot_grpc_dot_dummy__pb2.DummyRequest.FromString,
                    response_serializer=testproto_dot_grpc_dot_dummy__pb2.DummyReply.SerializeToString,
            ),
    }
    generic_handler = grpc.method_handlers_generic_handler(
            'dummy.DummyService', rpc_method_handlers)
    server.add_generic_rpc_handlers((generic_handler,))


 # This class is part of an EXPERIMENTAL API.
class DummyService(object):
    """DummyService
    """

    @staticmethod
    def UnaryUnary(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_unary(request, target, '/dummy.DummyService/UnaryUnary',
            testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
            testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
            options, channel_credentials,
            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

    @staticmethod
    def UnaryStream(request,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.unary_stream(request, target, '/dummy.DummyService/UnaryStream',
            testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
            testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
            options, channel_credentials,
            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

    @staticmethod
    def StreamUnary(request_iterator,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.stream_unary(request_iterator, target, '/dummy.DummyService/StreamUnary',
            testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
            testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
            options, channel_credentials,
            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

    @staticmethod
    def StreamStream(request_iterator,
            target,
            options=(),
            channel_credentials=None,
            call_credentials=None,
            insecure=False,
            compression=None,
            wait_for_ready=None,
            timeout=None,
            metadata=None):
        return grpc.experimental.stream_stream(request_iterator, target, '/dummy.DummyService/StreamStream',
            testproto_dot_grpc_dot_dummy__pb2.DummyRequest.SerializeToString,
            testproto_dot_grpc_dot_dummy__pb2.DummyReply.FromString,
            options, channel_credentials,
            insecure, call_credentials, compression, wait_for_ready, timeout, metadata)

dummy_pb2_grpc.pyi

"""
@generated by mypy-protobuf.  Do not edit manually!
isort:skip_file
https://github.com/vmagamedov/grpclib/blob/master/tests/dummy.proto"""
import abc
import grpc
import grpc.aio
import dummy_pb2
import typing

_T = typing.TypeVar('_T')

class _MaybeAsyncIterator(typing.AsyncIterator[_T], typing.Iterator[_T], metaclass=abc.ABCMeta):
    ...

class _ServicerContext(grpc.ServicerContext, grpc.aio.ServicerContext):  # type: ignore
    ...

class DummyServiceStub:
    """DummyService"""

    def __init__(self, channel: typing.Union[grpc.Channel, grpc.aio.Channel]) -> None: ...
    UnaryUnary: grpc.UnaryUnaryMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """UnaryUnary"""
    UnaryStream: grpc.UnaryStreamMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """UnaryStream"""
    StreamUnary: grpc.StreamUnaryMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """StreamUnary"""
    StreamStream: grpc.StreamStreamMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """StreamStream"""

class DummyServiceAsyncStub:
    """DummyService"""

    UnaryUnary: grpc.aio.UnaryUnaryMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """UnaryUnary"""
    UnaryStream: grpc.aio.UnaryStreamMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """UnaryStream"""
    StreamUnary: grpc.aio.StreamUnaryMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """StreamUnary"""
    StreamStream: grpc.aio.StreamStreamMultiCallable[
        dummy_pb2.DummyRequest,
        dummy_pb2.DummyReply,
    ]
    """StreamStream"""

class DummyServiceServicer(metaclass=abc.ABCMeta):
    """DummyService"""

    @abc.abstractmethod
    def UnaryUnary(
        self,
        request: dummy_pb2.DummyRequest,
        context: _ServicerContext,
    ) -> typing.Union[dummy_pb2.DummyReply, typing.Awaitable[dummy_pb2.DummyReply]]:
        """UnaryUnary"""
    @abc.abstractmethod
    def UnaryStream(
        self,
        request: dummy_pb2.DummyRequest,
        context: _ServicerContext,
    ) -> typing.Union[typing.Iterator[dummy_pb2.DummyReply], typing.AsyncIterator[dummy_pb2.DummyReply]]:
        """UnaryStream"""
    @abc.abstractmethod
    def StreamUnary(
        self,
        request_iterator: _MaybeAsyncIterator[dummy_pb2.DummyRequest],
        context: _ServicerContext,
    ) -> typing.Union[dummy_pb2.DummyReply, typing.Awaitable[dummy_pb2.DummyReply]]:
        """StreamUnary"""
    @abc.abstractmethod
    def StreamStream(
        self,
        request_iterator: _MaybeAsyncIterator[dummy_pb2.DummyRequest],
        context: _ServicerContext,
    ) -> typing.Union[typing.Iterator[dummy_pb2.DummyReply], typing.AsyncIterator[dummy_pb2.DummyReply]]:
        """StreamStream"""

def add_DummyServiceServicer_to_server(servicer: DummyServiceServicer, server: typing.Union[grpc.Server, grpc.aio.Server]) -> None: ...

dummy_pb2.py

# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler.  DO NOT EDIT!
# source: testproto/grpc/dummy.proto
"""Generated protocol buffer code."""
from google.protobuf.internal import builder as _builder
from google.protobuf import descriptor as _descriptor
from google.protobuf import descriptor_pool as _descriptor_pool
from google.protobuf import symbol_database as _symbol_database
# @@protoc_insertion_point(imports)

_sym_db = _symbol_database.Default()




DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x1atestproto/grpc/dummy.proto\x12\x05\x64ummy\"\x1d\n\x0c\x44ummyRequest\x12\r\n\x05value\x18\x01 \x01(\t\"\x1b\n\nDummyReply\x12\r\n\x05value\x18\x01 \x01(\t2\xfa\x01\n\x0c\x44ummyService\x12\x36\n\nUnaryUnary\x12\x13.dummy.DummyRequest\x1a\x11.dummy.DummyReply\"\x00\x12\x39\n\x0bUnaryStream\x12\x13.dummy.DummyRequest\x1a\x11.dummy.DummyReply\"\x00\x30\x01\x12\x39\n\x0bStreamUnary\x12\x13.dummy.DummyRequest\x1a\x11.dummy.DummyReply\"\x00(\x01\x12<\n\x0cStreamStream\x12\x13.dummy.DummyRequest\x1a\x11.dummy.DummyReply\"\x00(\x01\x30\x01\x62\x06proto3')

_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, globals())
_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'dummy_pb2', globals())
if _descriptor._USE_C_DESCRIPTORS == False:

  DESCRIPTOR._options = None
  _DUMMYREQUEST._serialized_start=37
  _DUMMYREQUEST._serialized_end=66
  _DUMMYREPLY._serialized_start=68
  _DUMMYREPLY._serialized_end=95
  _DUMMYSERVICE._serialized_start=98
  _DUMMYSERVICE._serialized_end=348
# @@protoc_insertion_point(module_scope)

dummy_pb2.pyi

"""
@generated by mypy-protobuf.  Do not edit manually!
isort:skip_file
https://github.com/vmagamedov/grpclib/blob/master/tests/dummy.proto"""
import builtins
import google.protobuf.descriptor
import google.protobuf.message
import sys

if sys.version_info >= (3, 8):
    import typing as typing_extensions
else:
    import typing_extensions

DESCRIPTOR: google.protobuf.descriptor.FileDescriptor

@typing_extensions.final
class DummyRequest(google.protobuf.message.Message):
    DESCRIPTOR: google.protobuf.descriptor.Descriptor

    VALUE_FIELD_NUMBER: builtins.int
    value: builtins.str
    def __init__(
        self,
        *,
        value: builtins.str = ...,
    ) -> None: ...
    def ClearField(self, field_name: typing_extensions.Literal["value", b"value"]) -> None: ...

global___DummyRequest = DummyRequest

@typing_extensions.final
class DummyReply(google.protobuf.message.Message):
    DESCRIPTOR: google.protobuf.descriptor.Descriptor

    VALUE_FIELD_NUMBER: builtins.int
    value: builtins.str
    def __init__(
        self,
        *,
        value: builtins.str = ...,
    ) -> None: ...
    def ClearField(self, field_name: typing_extensions.Literal["value", b"value"]) -> None: ...

global___DummyReply = DummyReply

test_grpc_async_usage.py

import typing
import pytest

import grpc
import dummy_pb2, dummy_pb2_grpc

ADDRESS = "localhost:22223"


class Servicer(dummy_pb2_grpc.DummyServiceServicer):
    async def UnaryUnary(
        self,
        request: dummy_pb2.DummyRequest,
        context: grpc.aio.ServicerContext,
    ) -> dummy_pb2.DummyReply:
        return dummy_pb2.DummyReply(value=request.value[::-1])

    async def UnaryStream(
        self,
        request: dummy_pb2.DummyRequest,
        context: grpc.aio.ServicerContext,
    ) -> typing.AsyncIterator[dummy_pb2.DummyReply]:
        for char in request.value:
            yield dummy_pb2.DummyReply(value=char)

    async def StreamUnary(
        self,
        request: typing.AsyncIterator[dummy_pb2.DummyRequest],
        context: grpc.aio.ServicerContext,
    ) -> dummy_pb2.DummyReply:
        values = [data.value async for data in request]
        return dummy_pb2.DummyReply(value="".join(values))

    async def StreamStream(
        self,
        request: typing.AsyncIterator[dummy_pb2.DummyRequest],
        context: grpc.ServicerContext,
    ) -> typing.AsyncIterator[dummy_pb2.DummyReply]:
        async for data in request:
            yield dummy_pb2.DummyReply(value=data.value.upper())


def make_server() -> grpc.aio.Server:
    server = grpc.aio.server()
    servicer = Servicer()
    server.add_insecure_port(ADDRESS)
    dummy_pb2_grpc.add_DummyServiceServicer_to_server(servicer, server)
    return server


@pytest.mark.asyncio
async def test_grpc() -> None:
    server = make_server()
    await server.start()
    async with grpc.aio.insecure_channel(ADDRESS) as channel:
        client: dummy_pb2_grpc.DummyServiceAsyncStub = dummy_pb2_grpc.DummyServiceStub(channel)  # type: ignore
        request = dummy_pb2.DummyRequest(value="cprg")
        result1 = await client.UnaryUnary(request)
        result2 = client.UnaryStream(dummy_pb2.DummyRequest(value=result1.value))
        result2_list = [r async for r in result2]
        assert len(result2_list) == 4
        result3 = client.StreamStream(dummy_pb2.DummyRequest(value=part.value) for part in result2_list)
        result3_list = [r async for r in result3]
        assert len(result3_list) == 4
        result4 = await client.StreamUnary(dummy_pb2.DummyRequest(value=part.value) for part in result3_list)
        assert result4.value == "GRPC"

    await server.stop(None)

run.sh

#!/usr/bin/env bash
set -o errexit -o nounset -o pipefail
python -m venv venv
source ./venv/bin/activate
pip install \
    protobuf==4.21.9 \
    pytest==7.2.0 \
    pytest-asyncio==0.20.3 \
    grpcio-tools==1.50.0 \
    mypy \
    types-protobuf==4.21.0.1

python -m pytest test_grpc_async_usage.py
mypy test_grpc_async_usage.py

Checklist:

• I have verified my MRE is sufficient to demonstrate the issue and solution by attempting to execute it myself
• I have added tests to typesafety/test_grpc.yml for all APIs added or changed by this PR
• I have removed any code generation notices from anything seeded using mypy-protobuf.

[RobinMcCorkell]

@shabbyrobe any thoughts on this PR? The tests should provide the Minimal Reproducible Examples that you need to evaluate the contribution.

[shabbyrobe]

Thank you for the contribution, and than you for the reminder.

I suspect I need to clarify the language about minimal reproducible examples: they're so I can run code that exercises a contribution locally, myself. There's really no substitute for that, especially for major changes. The typing tests are great, but they aren't enough.

Given I've had to revert disruptive aio typings PRs in the past, I'm afraid I can't accept this PR without one.

[shabbyrobe]

Looks good! Just a few things to address around FIXMEs and typing.Any. With those minor fixes and a proper MRE, I'd be happy to merge this in. Thank you again!

[RobinMcCorkell]

Added an MRE (basically a direct copy of the test from nipunn1313/mypy-protobuf#489). Looking at your comments.

[RobinMcCorkell]

I have an idea: what about replacing Any here (and in other partially typed signatures) with an uninhabited type? Then all usage of that type would need an explicit ignore, which punts safety to the user and allows these stubs to evolve in the correct direction later. Wdyt?

…

On Sat, 4 Mar 2023, 10:20 Blake Williams, ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In grpc-stubs/aio.pyi <#33 (comment)>: > +from concurrent import futures +from types import TracebackType +from . import ( + _Options, + Compression, + GenericRpcHandler, + ServerCredentials, + StatusCode, + ChannelCredentials, + CallCredentials, +) + +"""Create Client""" + +# FIXME +ClientInterceptor = typing.Any I think I take a different default stance on typing.Any. If we get it wrong, it risks introducing a breaking change in the typings later. If we leave it typing.Any, it guarantees introducing a breaking change in the typings later. I need more convincing here. — Reply to this email directly, view it on GitHub <#33 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAPMM3XKNVVVZXK6H3CVNMTW2KDDFANCNFSM6AAAAAAUWM6BYQ> . You are receiving this because you authored the thread.Message ID: ***@***.***>

[shabbyrobe]

I have an idea: what about replacing Any here (and in other partially typed signatures) with an uninhabited type?

That sounds like it could solve both our problems! How would that look?

[shabbyrobe]

Looks good! Let's go with it. Thank you for the quick turnaround for those tweaks, and thank you for the MRE, it will help a lot if any issues arise.

[shabbyrobe]

Just about to release to pypi, wondering how we version this. It could be breaking for folks consuming it but expecting aio to be untyped. I was originally tracking the grpc version itself, but I wasn't super disciplined there and became unmoored from that some time ago.

Split versioning of separate types and the original library is, to my mind, an unsolved problem. I still ultimately want it to be clear which version of grpc the typings are based on. The typings need a major version and a patch version too, with semver-lite semantics.

I wonder if maybe it needs to be versioned like this: <typing-major>.<library-major>.<library-minor>.<library-patch>.<typing-patch>. It's a bit long-winded but it could do the trick.

[RobinMcCorkell]

I think the major and minor version of grpc-stubs should match the grpc package. IMHO a separate grpc-stubs "major" version isn't necessary, since if there were any future massive type changes (although not sure how that would play out while still being compatible with grpc) then I think a whole new package/fork would be better than a bumped major version.

Keep a patch version independent of grpc for fixes in the stubs, but honestly I think that's sufficient control here.

As for the unmoored version of grpc vs grpc-stubs, I'm not sure it matters too much. A user will see grpc-stubs has an older minor version, and will likely deduce that some new features in grpc will be incorrectly typed but existing features will remain compatible (as compatible as type stubs can be at least). I would update the minor version of grpc-stubs to match grpc each time grpc-stubs releases, i.e. the minor version will likely be sparse and have gaps.

[shabbyrobe]

Thanks for you perspective. I'll skip on the "typing major" concept for now but I'll keep it in my pocket. There are still some problems I think it has the potential to solve, but I'll wait until I've seen instances of them happen in practice first.

I will go with a sub-patch version for typing updates from now on though, and I'll publish now. Thanks again for your contribution, it has been a gap for a while.

I think the major and minor version of grpc-stubs should match the grpc package.

They should, but it needs a once-over to make sure it still lines up with the docs.