技術共有

한어Русский языкEnglishFrançaisIndonesianSanskrit日本語DeutschPortuguêsΕλληνικάespañolItalianoSuomalainenLatina

Python Flask および gRPC サンプル プロジェクト

この記事では、PythonでFlaskとgRPCを使った簡単なサンプルアプリケーションを作成し、利用する方法を紹介します。 requests テスト用のライブラリ。

環境設定

まず、Python がインストールされていることを確認してください。次に、依存関係を管理するための仮想環境を作成します。

python -m venv myenv
source myenv/bin/activate  # Windows 使用 `myenvScriptsactivate`
1
2

必要なパッケージをインストールします。

pip install Flask grpcio grpcio-tools requests
1

gRPC サービスを定義する

作成する .proto gRPC サービスを定義するファイル。保存ファイル名service.proto：

syntax = "proto3";

package demo;

service DemoService {
  rpc SayHello (HelloRequest) returns (HelloResponse) {}
}

message HelloRequest {
  string name = 1;
}

message HelloResponse {
  string message = 1;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

Proto ファイルから Python コードを生成する

使用 grpc_tools Python コードを生成します。

python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. service.proto
1

これにより、次の 2 つのファイルが生成されます。service_pb2.py そしてservice_pb2_grpc.py。

gRPCサーバーを実装する

というファイルを作成します grpc_server.py 書類：

from concurrent import futures
import grpc
import service_pb2
import service_pb2_grpc

class DemoService(service_pb2_grpc.DemoServiceServicer):
    def SayHello(self, request, context):
        return service_pb2.HelloResponse(message=f'Hello, {request.name}!')

def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    service_pb2_grpc.add_DemoServiceServicer_to_server(DemoService(), server)
    server.add_insecure_port('[::]:50051')
    server.start()
    print("gRPC server started on port 50051")
    server.wait_for_termination()

if __name__ == '__main__':
    serve()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

Flaskサーバーを実装する

というファイルを作成します flask_server.py 書類：

from flask import Flask, request, jsonify
import grpc
import service_pb2
import service_pb2_grpc

app = Flask(__name__)

def get_grpc_client():
    channel = grpc.insecure_channel('localhost:50051')
    stub = service_pb2_grpc.DemoServiceStub(channel)
    return stub

@app.route('/hello', methods=['POST'])
def say_hello():
    data = request.json
    name = data.get('name')
    stub = get_grpc_client()
    response = stub.SayHello(service_pb2.HelloRequest(name=name))
    return jsonify({'message': response.message})

if __name__ == '__main__':
    app.run(port=5000, debug=True)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

サーバーを実行する

2 つのターミナル ウィンドウまたはタブを開きます。

1. 最初のターミナルで gRPC サーバーを起動します。

   python grpc_server.py
   1

2. 2 番目のターミナルで Flask サーバーを起動します。

   python flask_server.py
   1

使用 requests テストを実施する

というファイルを作成します test_flask_server.py ファイルを開き、次のコードを追加します。

import requests

def test_say_hello():
    url = 'http://localhost:5000/hello'
    data = {'name': 'World'}
    response = requests.post(url, json=data)
    
    if response.status_code == 200:
        print('Success!')
        print('Response JSON:', response.json())
    else:
        print('Failed!')
        print('Status Code:', response.status_code)
        print('Response Text:', response.text)

if __name__ == '__main__':
    test_say_hello()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

テストの実行

1. 確認する grpc_server.py そしてflask_server.py ランニング。

2. 別のターミナルで実行する test_flask_server.py：

   python test_flask_server.py
   1

次のような出力が表示されるはずです。

Success!
Response JSON: {'message': 'Hello, World!'}
1
2

スクリーンショットを実行する

このテスト スクリプトは、次の内容のメッセージを送信します。 name のために"World" JSONオブジェクトのPOSTリクエストhttp://localhost:5000/helloをクリックし、サーバーから返された応答を出力します。サーバーが正常に動作している場合は、成功メッセージが表示され、JSON データが返されます。