The project generates async gRPC stubs compatible with grpclib.
Client Implementation:
Use the generated *Stub class. Pass a grpclib.client.Channel to the stub. Remember to close the channel when finished.
Server Implementation:
Subclass the generated *Base class (e.g., EchoBase) and override the service methods. Use grpclib.server.Server to run the service.
Note: Async gRPC stub generation is enabled by default and provides improved static type checking and code completion.
# Client Example
import asyncio
import echo
from grpclib.client import Channel
async def main():
channel = Channel(host="127.0.0.1", port=50051)
service = echo.EchoStub(channel)
response = await service.echo(echo.EchoRequest(value="hello", extra_times=1))
print(response)
channel.close()
# Server Example
import asyncio
from echo import EchoBase, EchoRequest, EchoResponse, EchoStreamResponse
from grpclib.server import Server
from typing import AsyncIterator
class EchoService(EchoBase):
async def echo(self, echo_request: "EchoRequest") -> "EchoResponse":
return EchoResponse([echo_request.value for _ in range(echo_request.extra_times)])
async def echo_stream(self, echo_request: "EchoRequest") -> AsyncIterator["EchoStreamResponse"]:
for _ in range(echo_request.extra_times):
yield EchoStreamResponse(echo_request.value)
async def main():
server = Server([EchoService()])
await server.start("127.0.0.1", 50051)
await server.wait_closed()
