概要
最近少し話題になっている「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
- Request :
- Todoの一覧表示
- Request :
- Response :
- Todo[] : Todo一覧
- Todoのアップデート
- Request :
- string : Todo名
- boolean : 完了ならTrue
- Response :
- boolean : 成功ならTrue
- string : message
- Request :
蛇足
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での開発の流れは
- protoにAPIのエンドポイントとパラメータを決定
- コンパイル
- 自動生成されたクラスを使って本体開発
の3ステップだけ
接続方法を自動生成してくれるため非常に開発が簡単になりましたね！
おしまい。