POC for using betterproto with grpcio without generated a servicer

bp_grcpio_example_usage.py

from magic_glue import create_grpc_handler, rpc_method
from .generated_code.mypackage import MyRequest, MyResponse

# Create a servicer class with the rpc methods we want to implement
class MyServicer:

    # grpcio needs to know the fully qualified name of the service
    # It would be better if this was available from the betterproto generated classes
    # to save having to copy it from the proto file manually
    service_name = "mypackage.MyService"

    @rpc_method()
    def MyMethod(self, request: MyRequest, context) -> MyResponse:
        # the method name has to match the protobuf rpc name, or provide name kwarg to 
        # the decorator instead if you can't stand camelcase
        return MyResponse(...)


# Create grpcio server
server = grpc.server()

# Create a handler and register it with the server
server.add_generic_rpc_handlers((create_grpc_handler(MyServicer()), ))

# And you can call/test it
MyServiceStub = create_stub_class(MyServicer)
test_client = MyServiceStub(test_channel)
response = test_client.MyMethod(MyRequest(...))

magic_glue.py

"""
This module provides basically everything you need to make it convenient to use 
of betterproto classes with grpcio without any extra generated servicer code.
"""

def rpc_method(*, request=None, response=None, nam e=None):
    """
    A decorator to apply to servicer methods to annotate them as rpc
    methods for the handler.

    TODO: extend this to allow specification of stream io
    """

    def inner_decorator(func):
        signature = typing.get_type_hints(func)
        func.__rpc_method__ = {
            "request": signature.get("request", request),
            "response": signature.get("return", response),
            "name": name or func.__name__,
        }
        assert func.__rpc_method__["request"], "request type must be annotated"
        assert func.__rpc_method__["response"], "response type must be annotated"
        return func

    return inner_decorator


def is_rpc_method(value):
    """
    Check if the given value is a function annotated with __rpc_method__
    """
    return callable(value) and isinstance(getattr(value, "__rpc_method__", None), dict)


def create_stub_class(servicer_cls):
    """
    Creates a grpcio testing client for the given servicer_cls
    """

    class GenericStub:
        def __init__(self, channel):
            rpc_methods = inspect.getmembers(servicer_cls, is_rpc_method)
            for name, method in rpc_methods:
                method_stub = channel.unary_unary(
                    f"/{servicer_cls.service_name}/{method.__rpc_method__.get('name')}",
                    request_serializer=method.__rpc_method__.get(
                        "request"
                    ).SerializeToString,
                    response_deserializer=method.__rpc_method__.get(
                        "response"
                    ).FromString,
                )
                setattr(self, name, method_stub)

    return GenericStub


def create_grpc_handler(servicer):
    """
    Create a handler from the given servicer class that can be added to a grpcio
    server.

    TODO: add support for stream methods in the decorator
    """
    return grpc.method_handlers_generic_handler(
        servicer.service_name,
        {
            method.__rpc_method__.get("name"): grpc.unary_unary_rpc_method_handler(
                method,
                request_deserializer=getattr(
                    method.__rpc_method__.get("request"), "FromString", None
                ),
                response_serializer=getattr(
                    method.__rpc_method__.get("response"), "SerializeToString", None
                ),
            )
            for _, method in inspect.getmembers(servicer, is_rpc_method)
        },
    )