Add grpcio base generation support

Author: roblabla

src/betterproto/plugin/models.py

@@ -563,6 +563,10 @@ def __post_init__(self) -> None:
     def proto_name(self) -> str:
         return self.proto_obj.name
 
+    @property
+    def proto_path(self) -> str:
+        return self.parent.package + "." + self.proto_name
+
     @property
     def py_name(self) -> str:
         return pythonize_class_name(self.proto_name)

src/betterproto/templates/template.py.j2

@@ -21,6 +21,9 @@ from betterproto.grpc.grpclib_server import ServiceBase
 {% if output_file.services and output_file.parent_request.options.grpc_kind == "grpclib" %}
 import grpclib
 {% endif %}
+{% if output_file.services and output_file.parent_request.options.grpc_kind == "grpcio" %}
+import grpc
+{% endif %}
 
 
 {% if output_file.enums %}{% for enum in output_file.enums %}
@@ -244,6 +247,60 @@ class {{ service.py_name }}Base(ServiceBase):
 {% endfor %}
 {% endif %}
 
+{% if output_file.parent_request.options.grpc_kind == "grpcio" %}
+{% for service in output_file.services %}
+class {{ service.py_name }}Base:
+    {% if service.comment %}
+{{ service.comment }}
+
+    {% endif %}
+
+    {% for method in service.methods %}
+    async def {{ method.py_name }}(self
+        {%- if not method.client_streaming -%}
+            , request: "{{ method.py_input_message_type }}"
+        {%- else -%}
+            {# Client streaming: need a request iterator instead #}
+            , request_iterator: AsyncIterator["{{ method.py_input_message_type }}"]
+        {%- endif -%}
+            , context: grpc.aio.ServicerContext
+            ) -> {% if method.server_streaming %}AsyncGenerator["{{ method.py_output_message_type }}", None]{% else %}"{{ method.py_output_message_type }}"{% endif %}:
+        {% if method.comment %}
+{{ method.comment }}
+
+        {% endif %}
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    {% endfor %}
+
+    def register_to_server(self, server: grpc.aio.Server):
+        rpc_method_handlers = {
+        {% for method in service.methods %}
+            "{{ method.proto_name }}":
+            {% if not method.client_streaming and not method.server_streaming %}
+                grpc.unary_unary_rpc_method_handler(
+            {% elif method.client_streaming and method.server_streaming %}
+                grpc.stream_stream_rpc_method_handler(
+            {% elif method.client_streaming %}
+                grpc.stream_unary_rpc_method_handler(
+            {% else %}
+                grpc.unary_stream_rpc_method_handler(
+            {% endif %}
+                    self.{{ method.py_name }},
+                    request_deserializer={{ method.py_input_message_type }}.FromString,
+                    response_serializer={{ method.py_input_message_type }}.SerializeToString,
+                ),
+        {% endfor %}
+        }
+        generic_handler = grpc.method_handlers_generic_handler(
+            "{{ service.proto_path }}", rpc_method_handlers)
+        self.add_generic_rpc_handlers((generic_handler,))
+
+{% endfor %}
+{% endif %}
+
 {% for i in output_file.imports|sort %}
 {{ i }}
 {% endfor %}