{
	"1": {
		"id": 1,
		"nickname": "admin",
		"mail_address": "admin@example.com",
		"user_type": "ADMINISTRATOR"
	},
	"2": {
		"id": 2,
		"nickname": "guest",
		"mail_address": "guest@example.com",
		"user_type": "GUEST"
	},
	"3": {
		"id": 3,
		"nickname": "foo",
		"mail_address": "foo@example.com",
		"user_type": "NORMAL"
	},
	"4": {
		"id": 4,
		"nickname": "troll",
		"mail_address": "troll@example.com",
		"user_type": "DISABLED"
	},
	"5": {
		"id": 5,
		"nickname": "Zaru",
		"mail_address": "zaru@example.com",
		"user_type": "NORMAL"
	},
	"6": {
		"id": 6,
		"nickname": "Saru",
		"mail_address": "saru@example.com",
		"user_type": "NORMAL"
	}
}

# gRPC serverに登録するservicer
from servicers.user import UserManager
# gRPCのサーバー実装ではThreadPoolを利用する
from concurrent.futures import ThreadPoolExecutor

# 「grpc」パッケージと、grpc_tools.protocによって生成したパッケージをimportする
import grpc
import user_pb2
import user_pb2_grpc

# grpc reflection用の追加ライブラリ
from grpc_reflection.v1alpha import reflection


def manager():
    # Serverオブジェクトを作成する
    server = grpc.server(ThreadPoolExecutor(max_workers=2))

    # Serverオブジェクトに定義したServicerクラスを登録する
    user_pb2_grpc.add_UserManagerServicer_to_server(UserManager(), server)

    # [追記] リフレクション登録
    SERVICE_NAMES = (
        reflection.SERVICE_NAME,
    )
    SERVICE_NAMES += (user_pb2.DESCRIPTOR.services_by_name[UserManager.__name__].full_name,)
    reflection.enable_server_reflection(SERVICE_NAMES, server)

    # 1234番ポートで待ち受けするよう指定する
    server.add_insecure_port("[::]:1234")

    # 待ち受けを開始する
    server.start()

    # 待ち受け終了後の後処理を実行する
    server.wait_for_termination()

if __name__ == "__main__":
    manager()

from typing import List
from pprint import pprint
import sys

# 「grpc」パッケージと、protocによって生成したパッケージをimportする
import grpc
import user_pb2
import user_pb2_grpc


def main():
    # サーバーに接続する
    channel = grpc.insecure_channel("localhost:1234")
    # 送信先の「stub」を作成する
    stub = user_pb2_grpc.UserManagerStub(channel)
    # それぞれのリクエストを送信する
    if sys.argv[1] == "GetUser":
        res = get_user(stub=stub, user_id=int(sys.argv[2]))
    elif sys.argv[1] == "AddUser":
        res = add_user(
            stub=stub,
            user_id=int(sys.argv[2]),
            nickname=sys.argv[3],
            mail_address=sys.argv[4],
            user_type=int(sys.argv[5]),
        )
    elif sys.argv[1] == "CountAlreadyUsers":
        user_id_list = [int(sys.argv[2]), int(sys.argv[3])]
        res = count_already_users(stub=stub, user_id_list=user_id_list)
    elif sys.argv[1] == "GetUsersByType":
        res = get_users_by_type(stub=stub, user_type=int(sys.argv[2]))
    elif sys.argv[1] == "GetUsersByIds":
        user_id_list = [int(sys.argv[2]), int(sys.argv[3])]
        res = get_users_by_ids(stub=stub, user_id_list=user_id_list)
    return res


if __name__ == "__main__":
    main()

syntax = "proto3";

// ユーザー情報を表すメッセージ型
message User {
  uint32 id = 1;
  string nickname = 2;
  string mail_address = 3;
  enum UserType {
    NONE = 0;
    NORMAL = 1;
    ADMINISTRATOR = 2;
    GUEST = 3;
    DISABLED = 4;
  }
  UserType user_type = 4;
}

// ユーザーの情報リクエストに使用するメッセージ型
message UserRequest {
  uint32 id = 1;
}

// ユーザーの情報リクエストに使用するメッセージ型
message UserTypeRequest {
  enum UserType {
    NONE = 0;
    NORMAL = 1;
    ADMINISTRATOR = 2;
    GUEST = 3;
    DISABLED = 4;
  }
  UserType user_type = 4;
}

// ユーザー情報を返す際に使用するメッセージ型
message UserResponse {
  bool error = 1;
  string message = 2;
  User user = 3;
}

// ユーザー人数を返す際に使用するメッセージ型
message UserCntResponse {
  bool error = 1;
  uint32 user_cnt = 2;
}

// ユーザー管理を行うサービス
service UserManager {
  // ユーザー情報を取得する
  rpc GetUser (UserRequest) returns (UserResponse) {}
  // 新規のユーザー情報を追加する
  rpc AddUser (User) returns (UserResponse) {}
  // 与えられたユーザーのうち存在する人数を取得する
  rpc CountAlreadyUsers (stream UserRequest) returns (UserCntResponse) {}
  // 与えられたユーザータイプと同種のユーザーを取得する
  rpc GetUsersByType (UserTypeRequest) returns (stream UserResponse) {}
  rpc GetUsersByIds (stream UserRequest) returns (stream UserResponse) {}
}

class UserManager(user_pb2_grpc.UserManagerServicer):

    def GetUser(self, request: user_pb2.UserRequest, context):
        """
        Unary RPC(Simple RPC)
        ユーザー情報を取得する
        """
        # クライアントが送信した引数はrequest引数に格納され、
        # このオブジェクトに対しては一般的なPythonオブジェクトと
        # 同様の形でプロパティにアクセスできる
        user_id = request.id

        # ユーザー情報はユーザーIDを文字列に変換したものをキーとする辞書型データ
        if str(user_id) not in users:
            # 該当するユーザーが存在しない場合エラーを返す
            return user_pb2.UserResponse(error=True, message="not found")
        user = users[str(user_id)]

        # 戻り値として返すUserオブジェクトを作成する
        result = user_pb2.User(
            id=user["id"],
            nickname=user["nickname"],
            mail_address=user["mail_address"],
            user_type=user_pb2.User.UserType.Value(user["user_type"]),
        )
        print("result.user_type", result.user_type, type(result.user_type))  # debug

        # UserResponseオブジェクトを返す
        return user_pb2.UserResponse(error=False, user=result)

class UserManager(user_pb2_grpc.UserManagerServicer):

    def AddUser(self, request: user_pb2.User, context):
        """
        Unary RPC(Simple RPC)
        新規にユーザー情報を登録する
        """
        # クライアントが送信した引数はrequest引数に格納され、
        # このオブジェクトに対しては一般的なPythonオブジェクトと
        # 同様の形でプロパティにアクセスできる
        user_id = request.id

        # ユーザー情報はユーザーIDを文字列に変換したものをキーとする辞書型データ
        # なので、適宜文字列型に変換して使用している
        if str(user_id) in users:
            # 該当するユーザーが既に存在している場合エラーを返す
            return user_pb2.UserResponse(error=True, message="already exist")

        # 新規登録用及び戻り値として返すUserオブジェクトを作成する
        result = user_pb2.User(
            id=request.id,
            nickname=request.nickname,
            mail_address=request.mail_address,
            user_type=request.user_type,
        )
        # preserving_proto_field_nameでcamelCaseがsnake_caseに変換される
        users[str(request.id)] = json_format.MessageToDict(
            result, preserving_proto_field_name=True
        )
        with open(USER_DB, mode="w") as f:
            json.dump(users, f)

        # UserResponseオブジェクトを返す
        return user_pb2.UserResponse(error=False, user=result)

class UserManager(user_pb2_grpc.UserManagerServicer):

    def GetUsersByType(self, request: user_pb2.UserTypeRequest, context):
        """
        Server streaming RPC(Response-streaming RPC)
        渡されたユーザータイプと同じユーザーを複数返す
        """
        user_id_list = [u for u in users]
        print("user_id_list", user_id_list)  # debug
        result_list = []
        for u_id in user_id_list:
            user = users[u_id]
            # Name(数字), Value(NORMAL)とか
            if user_pb2.User.UserType.Name(request.user_type) == user["user_type"]:
                print(f"{user['id']}---該当あり---")
                result = user_pb2.User(
                    id=user["id"],
                    nickname=user["nickname"],
                    mail_address=user["mail_address"],
                    user_type=user_pb2.User.UserType.Value(user["user_type"]),
                )
                yield user_pb2.UserResponse(error=False, user=result)
        # 何も返さない条件が来てもエラーにならない

def get_users_by_type(stub, user_type: int):
    # リクエストに使用するオブジェクト（ここでは「UserTypeRequest」型オブジェクト）を作成
    req = user_pb2.UserTypeRequest(user_type=user_type)
    responses = stub.GetUsersByType(req)
    response_list = []
    for r in responses:
        # 取得したレスポンスの表示
        pprint(r)  # debug
        response_list.append(r)
    return response_list

    def CountAlreadyUsers(self, request_iter: Iterable[user_pb2.UserRequest], context):
        """
        Client streaming RPC(Request-streaming RPC)
        複数渡されたユーザーidのうち既に存在している人数を取得する
        """
        user_cnt = 0
        for request in request_iter:
            user_id = request.id
            # ユーザー情報はユーザーIDを文字列に変換したものをキーとする辞書型データ
            if str(user_id) in users:
                # 該当するユーザーが存在するならカウント
                user_cnt += 1

def count_already_users(stub, user_id_list: List[int]):
    # リクエストに使用するオブジェクト（ここでは「UserRequest」型オブジェクト）を複数作成
    req_list = []
    for user_id in user_id_list:
        req = user_pb2.UserRequest(id=user_id)
        req_list.append(req)
    response = stub.CountAlreadyUsers(iter(req_list))
    # 取得したレスポンスの表示
    pprint(response)
    return response

    def GetUsersByIds(self, request_iter: Iterable[user_pb2.UserRequest], context):
        """
        Bidrectional streaming RPC
        複数渡されたユーザーidのうち存在しているUserを返す
        """
        user_list = []
        for request in request_iter:
            user_id = request.id
            # ユーザー情報はユーザーIDを文字列に変換したものをキーとする辞書型データ
            if str(user_id) in users:
                user = users[str(user_id)]
                user_list.append(user)
        for user in user_list:
            result = user_pb2.User(
                id=user["id"],
                nickname=user["nickname"],
                mail_address=user["mail_address"],
                user_type=user_pb2.User.UserType.Value(user["user_type"]),
            )
            yield user_pb2.UserResponse(error=False, user=result)

def get_users_by_ids(stub, user_id_list: List[int]):
    # リクエストに使用するオブジェクト（ここでは「UserRequest」型オブジェクト）を複数作成
    req_list = []
    for user_id in user_id_list:
        req = user_pb2.UserRequest(id=user_id)
        req_list.append(req)
    responses = stub.GetUsersByIds(iter(req_list))
    response_list = []
    for r in responses:
        # 取得したレスポンスの表示
        pprint(r)  # debug
        response_list.append(r)
    return response_list