「Python×gRPC」で簡単！サーバー開発(Todoアプリ)

概要

最近少し話題になっている「gRPC」
APIの開発が簡単になる次世代的なモノ、と聞いたことは多いのではないでしょうか？(自分はそのひとり)

Python×gRPCで簡単なTodoアプリの作成を通して、gRPCの使い方を紹介します。

gRPC #とは

「gRPCが何なのか」の解説は先人の皆様が解説してくださっているので、ここでは「Python×gRPCでの開発例」について話します。

要点は以下

• API設計をソースコードに落とし込める(=*.proto)
• *.protoからAPIの実装に必要なクラスを自動生成する
• 自動生成されたクラスが「よしなに」データをシリアライズ・デシリアライズしてくれる

準備

環境

今回は以下の環境で開発を進めます。ただ個々人で環境は違うと思うので、必須の項目だけ太文字にしておきます。

• OS : Windows
• pip : 最新
• Python : Python3.x
• 関心 : 強く持つ

また完成物は以下にあげておきます。早くブツが見たい方はどうぞ。

Github : python-gRPC

インストール

*.protoファイルをコンパイルするために、grpcioをインストールしておきましょう。

pip install grpcio-tools

Todoアプリの仕様

作成するTodoアプリのAPI仕様は以下。

• 共通
  • Request : string timestamp
  • Response : string timestamp
• Todoの登録
  • Request :
    • string : Todo名
  • Response :
    • boolean : 成功ならTrue
    • string : message
• Todoの一覧表示
  • Request :
  • Response :
    • Todo[] : Todo一覧
• Todoのアップデート
  • Request :
    • string : Todo名
    • boolean : 完了ならTrue
  • Response :
    • boolean : 成功ならTrue
    • string : message

蛇足

REST APIであれば、先のgRPC APIは以下のような感じでしょう……？

• Todoの登録(POST /todo)
• Todoの一覧表示(GET /todo)
• Todoのアップデート(PUT /todo/{todo_name})

開発

proto(ProtocolBuffers)

実際にコーディングしていきます。まずは上記のAPIを実際にコードに起こしておきましょう。

route.proto

// proto3記法であることを表している
syntax = "proto3";

package todo;

/*
message命名規則(ベストプラクティスはどんなものだろう……)
*Component : Request/Responseに用いるコンポーネント
Todo[Action]Request : Action用リスエスト内容
Todo[Action]Response : Action用レスポンス内容
*/

// TODO
message TodoComponent {
    string todo_name = 1;
    bool is_done = 2;
}

// サーバーからのレスポンス(共通)
message ServerResponseComponent {
    bool is_success = 1;
    string message = 2;
}

// TODOを作成するときのリクエスト
message TodoCreateRequest {
    string todo_name = 1;
    string timestamp = 3;
}

// TODOを作成するときのレスポンス
// messageの中に別のmessageを含むことができる
message TodoCreateResponse {
    ServerResponseComponent response = 1;
    string timestamp = 2;
}


message TodoShowRequest {
    string timestamp = 1;
}

// repeated SOMETHING = SOMETHING[]
message TodoShowResponse {
    repeated TodoComponent todos = 1;
    string timestamp = 3;
}

message TodoUpdateRequest {
    TodoComponent todo = 1;
    string timestamp = 2;
}

message TodoUpdateResponse {
    ServerResponseComponent response = 1;
    string timestamp = 2;
}

// ルーティング
service TodoGateway {
    rpc TodoCreate (TodoCreateRequest) returns (TodoCreateResponse) {}
    rpc TodoShow (TodoShowRequest) returns (TodoShowResponse) {};
    rpc TodoUpdate (TodoUpdateRequest) returns (TodoUpdateResponse) {}
}

protoの簡単な解説

message

APIの送受信データ。
コンパイル(後述)するとリクエスト・レスポンス用のクラスになる。

実際に通信をするときにはこのmessageにしたがってデータのシリアライズ・デシリアライズが行われる模様。

sample.proto

message MessageName {
    型 変数名 = 固有ID;
}

で記述する。
また、message内にバリデーションを定義することもできる。

service

APIのルーティング部分。
コンパイル(後述)するとサーバーの抽象クラスになる。

sample.proto

service ServiceName {
    rpc 名前 (受け取るmessage) returns 返却するmessage;
}

で記述する。

protoのコンパイルコード

protoをPythonコンパイルするには

$ python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. ./route.proto

とシェルで入力するか

code_gen.py

from grpc.tools import protoc

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

を作成し、

$ python ./code_gen.py

としても良い。

上記いずれかを実行すると

• route_pb2.py
  • messageから生成されたクラス一覧
• route_pb2_grpc.py
  • serviceから生成されたサーバークラス

の2つのpythonファイルが生成される。

サーバー側の開発

サーバー側のPython実装は以下

TodoGatewayServicerを継承し、API用のメソッドを定義してあげるだけでOK
またリクエストデータは第2引数のrequestの中に含まれている。

from route_pb2 import *
from route_pb2_grpc import add_TodoGatewayServicer_to_server, TodoGatewayServicer

from concurrent import futures
from datetime import datetime
import time
import grpc


_ONE_DAY_IN_SECONDS = 60 * 60 * 24


def get_timestamp():
    return datetime.now().strftime("%Y/%m/%d %H:%M:%S")


class RouteTodoServicer(TodoGatewayServicer):
    """
    Todoサーバーの実装
    protoで定義したservice名のメソッドを実装する
    """

    todos = {}

    # TODOを作成するロジック
    def TodoCreate(self, request, response):
        # request.fieldでprotoで定義した値にアクセスできる
        print("Todo Create Called : %s" % request.timestamp)
        try:
            self.todos[request.todo_name] = TodoComponent(
                todo_name=request.todo_name,
                is_done=False    # 作成時はTODO完了していないためFalse
            )
            is_success = True
            message = None
        except:
            import traceback
            traceback.print_exc()
            is_success = False
            message = "something happen"

        # レスポンスを返す時はreturnするだけで良い
        return TodoCreateResponse(
            response=ServerResponseComponent(
                is_success=is_success,
                message=message
            ),
            timestamp=get_timestamp()
        )

    def TodoShow(self, request, response):
        print("Todo Show Called : %s" % request.timestamp)
        try:
            todo_list = [
                todo for todo in self.todos.values()
            ]
        except:
            import traceback
            traceback.print_exc()
            todo_list = []
        return TodoShowResponse(todos=todo_list, timestamp=get_timestamp())

    def TodoUpdate(self, request, response):
        print("Todo Update Called : %s" % request.timestamp)
        try:
            if request.todo.todo_name in self.todos:
                message = None
                # 入れ子になっているfieldにもアクセスできる
                self.todos[request.todo.todo_name].is_done = request.todo.is_done
            else:
                message = "No such a todo"
            is_success = True
        except:
            is_success = False
            message = "something happen"
        return TodoUpdateResponse(
            response=ServerResponseComponent(
                is_success=is_success,
                message=message
            ),
            timestamp=get_timestamp()
        )

# サーバーの実行
def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    add_TodoGatewayServicer_to_server(
        RouteTodoServicer(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("Server Start!!")
    try:
        while True:
            time.sleep(_ONE_DAY_IN_SECONDS)
    except KeyboardInterrupt:
        server.stop(0)


if __name__ == '__main__':
    serve()

サーバー側は従来のREST APIのように仕様書とにらめっこすることが不要になり
サーバーのロジック開発に集中することができます。

クライアント側の開発

クライアント側の実装は以下。
接続先を指定してstubを作成し、そのstubに向けてAPIを投げます。

from route_pb2 import *
from route_pb2_grpc import TodoGatewayStub
import grpc

from datetime import datetime


def get_timestamp():
    return datetime.now().strftime("%Y/%m/%d %H:%M:%S")


# データを送信する関数(関数名は何でもよい)
def create_todo(stub, todo_name):

    # stubにはTodoGatewayが実装されたgRPCサーバーへのアクセス情報が入っている
    # TodoGatewayにはTodoCreateメソッドが実装されているはず
    response = stub.TodoCreate(TodoCreateRequest(
        todo_name=todo_name,
        timestamp=get_timestamp()
    ))

    if response.response.is_success:
        print("create success")
    else:
        print("Error : " + response.response.message)


def show_todos(stub):
    response = stub.TodoShow(
        TodoShowRequest(
            timestamp=get_timestamp()
        )
    )

    print("---- Todo Name : is done? ----")

    # レスポンスの中のTODOリストにアクセス
    for todo in response.todos:
        print("%s : %s" % (todo.todo_name, todo.is_done))

    print("")


def update_todo(stub, todo_name, is_done):
    response = stub.TodoUpdate(
        TodoUpdateRequest(
            todo=TodoComponent(
                todo_name=todo_name,
                is_done=is_done
            ),
            timestamp=get_timestamp()
        )
    )

    if response.response.is_success:
        print("update success")
        if response.response.message:
            print("message : %s" % response.response.message)
    else:
        print("Error : " + response.response.message)


if __name__ == '__main__':

    # localhost:50051にTodoリクエストを送る準備
    with grpc.insecure_channel('localhost:50051') as channel:
        stub = TodoGatewayStub(channel)

        """
        使用方法 : 
            コマンドに従ってstub宛にリクエストを送信する関数を呼び出す
            Todo登録
                c|create [todo_name]
            Todo確認
                s|show
            Todo更新
                u|update [todo_name] [y/n]
        """

        while True:
            command = input().split()
            if len(command) == 0:
                break
            if command[0] == "c" or command[0] == "create":
                # create todo
                try:
                    todo_name = command[1]
                except:
                    print("input todo name: ", end="")
                    todo_name = input()
                create_todo(stub, todo_name)
            elif command[0] == "s" or command[0] == "show":
                # show todos
                show_todos(stub)
            elif command[0] == "u" or command[0] == "update":
                # update todo
                try:
                    todo_name = command[1]
                except:
                    print("input todo name: ", end="")
                    todo_name = input()
                try:
                    is_done = command[2] == "y" or command[2] == "yes"
                except:
                    print("is done ? y/n: ", end="")
                    _is_done = input()
                    is_done = _is_done == "y" or _is_done == "yes"
                update_todo(stub, todo_name, is_done)
            else:
                print("input an illigal command, try again.")

APIを叩く <=> stubを叩く

の2つが同じ挙動をしてくれるため
クライアント側の実装もAPIの仕様書を眺める必要はありません。

実行

サーバーの実行は以下。

python ./todo_server.py

クライアントは別のコンソールを開いて以下を叩きます。

python ./todo_client.py

おしまい

gRPCでの開発の流れは

1. protoにAPIのエンドポイントとパラメータを決定
2. コンパイル
3. 自動生成されたクラスを使って本体開発

の3ステップだけ
接続方法を自動生成してくれるため非常に開発が簡単になりましたね！

おしまい。