J'ai essayé gRPC avec Python

$ brew install protobuf
$ pip install grpcio-tools

syntax = "proto3";

package gateway;

message ReplyRequest {
    string message = 1;
}

message ReplyResponse {
    string message = 1;
}

service AVRGateway {
    rpc Reply (ReplyRequest) returns (ReplyResponse) {}
}

from grpc.tools import protoc


protoc.main(
    (
        '',
        '-I.',
        '--python_out=.',
        '--grpc_python_out=.',
        './hoge.proto',
    )
)

$ python ./codegen.py

from concurrent import futures
import time

import grpc

from hoge_pb2 import ReplyResponse
from hoge_pb2_grpc import AVRGatewayServicer
from hoge_pb2_grpc import add_AVRGatewayServicer_to_server

_ONE_DAY_IN_SECONDS = 60 * 60 * 24


class RouteGuideServicer(AVRGatewayServicer):
    def Reply(self, request, context):
        print('reply!')
        return ReplyResponse(message='Hoge')


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    add_AVRGatewayServicer_to_server(
            RouteGuideServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    try:
        while True:
            time.sleep(_ONE_DAY_IN_SECONDS)
    except KeyboardInterrupt:
        server.stop(0)


if __name__ == '__main__':
    serve()

from __future__ import print_function
import grpc
import hoge_pb2
import hoge_pb2_grpc


def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = hoge_pb2_grpc.AVRGatewayStub(channel)
    response = stub.Reply(hoge_pb2.ReplyRequest(message='hoge'))
    print("Greeter client received: " + response.message)


if __name__ == '__main__':
    run()

Recommended Posts