grpcでC#とpythonの通信をする

はじめに

grpcでC#からpythonへ通信をする方法をまとめました。
リポジトリ：https://github.com/kz-000/grpc_example

プロジェクトの構成は以下のようになっています。
grpc_example
　├ api (protoファイル)
　├ client (C#)
　└ server (python)

API

今回はIDを受け取り、ユーザー名を返すような簡単なAPIを実装しようと思います。

//user.ptoro
syntax = "proto3";

package api;

message GetUserRequest {
    int32 Id= 1;
}

message GetUserResponse {
    string Name = 1;
}

service UserService{
    rpc GetUser (GetUserRequest) returns (GetUserResponse) {}
}

サーバー（python）

grpcに必要なライブラリをpipでインストールします。

pip install grpcio = 1.50.0 grpcio-tools = 1.50.0

protoファイルから通信に必要なコードを生成するプログラムを実装します。
下記codegen.pyを実行すると、src以下にuser_pb2_grpc.pyとuser_pb2.pyが生成されます。

#codegen.py
from grpc.tools import protoc
protoc.main(
    (
        '',
        '-I../api/', #protoファイルが入っているディレクトリ
        '--python_out=./src/', #コード(pb2)の出力ディレクトリ
        '--grpc_python_out=./src/', #コード(pb2_grpc)の出力ディレクトリ
        'user.proto', #コンパイルするprotoファイル
    )
)

次にuser_pb2、user_pb2_grpcを使ってサービスの実装を行います。
受け取ったIDに応じてユーザー名を返す関数（GetUser）を作成します。

GetUserはProtoファイルで定義した関数のため、GetUserRequestを受け取り、GetUserResponseを返すという仕様を満たす必要があります。

import user_pb2 as model
import user_pb2_grpc as service


class UserServicer(service.UserService):
    def __init__(self):
        ...

    def GetUser(self, request, context):
        match request.Id:
            case 1:
                return model.GetUserResponse(Name="taro")
            case 2:
                return model.GetUserResponse(Name="jiro")
            case _:
                return model.GetUserResponse(Name="not found")

最後にサーバの実装をします。
add_UserServiceServicer_to_server(UserServicer(), server)で作成したサービスを追加します。
server.add_insecure_port('localhost:50004')でポートを指定しserver.start()でサーバが起動します。

#main.py
from src.user_pb2_grpc import add_UserServiceServicer_to_server
from src.user_servicer import UserServicer
from concurrent import futures
import grpc

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))

add_UserServiceServicer_to_server(UserServicer(), server)
server.add_insecure_port('localhost:50004')
server.start()
print('server started')
server.wait_for_termination()

python側の実装は以上です。

クライアント(C#)

NugetからGrpc、Grpc.Tools、Google.Protobufをインストールします。

client.csprojを開き、ItemGroupにprotoファイルのパスを追加します。
ビルドするobj/Debug/net7.0/UserGrpc.csというファイルが生成されます。これを使用してクライアント側の実装をしていきます。

<ItemGroup>
    <!-- 追加 -->
	<Protobuf Include="..\api\user.proto" Link="api\user.proto"  GrpcServices="Client"/>

	<PackageReference Include="Google.Protobuf" Version="3.23.4" />
	<PackageReference Include="Grpc" Version="2.46.6" />
	...
</ItemGroup>

クライアントの実装では、UserServiceClientをラップした以下のようなクラスを実装します。。

//UserClient.cs
using Api;
using Grpc.Core;

public class UserClient
{
    Api.UserService.UserServiceClient client;

    public UserClient(string ip = "localhost", int port = 50004)
    {
        var channel = new Channel($"{ip}:{port}", ChannelCredentials.Insecure);
        client = new Api.UserService.UserServiceClient(channel);
    }

    public void GetUser(string id)
    {
        var response = client.GetUser(new() { Id = id });
        Console.WriteLine(response.Name);
    }
}

実装したクライアントを使って、通信できるか確認します。

//Program.cs
var client = new UserClient();
client.GetUser(1);
client.GetUser(2);
client.GetUser(3);

クライアントを実行する前に、サーバのプログラムを必ず実行してください。

以下のような出力が返ってくれば成功です。

taro
jiro
not found