Pythonでいろんなサーバーを立ててみる

「こんな感じのAPIサーバー立てといて」なんて言葉、開発してれば日常茶飯事です。 そんな「APIサーバーを立てる」と言ってもいろんな種類がありますね。

今回はよくあるAPIサーバーを一通りPythonで立てるだけ立ててみようと思います。

比較

最初に今回扱うサーバー3種類の紹介です。

• REST
• gRPC
• GraphQL

これらをざっくり比較したものを探すと下記のような表を見つけました。

	HTTPプロトコル	エンドポイント	パフォーマンス	データ型のサポート
REST API	HTTP/1.1、HTTP/2	リソースごとにエンドポイントが設けられることが多い。そのリソースに対する操作をGET, POST, PUT, DELETEなどで表現。	△(JSONであることや、オーバ/アンダーフェッチは起こり得る)	JSON、XML、MessagePackなど
gRPC	HTTP/2	サービスごとに.protoファイルで定義され、サービス内の各メソッドが実質的なエンドポイントとなる。	◯(ProtobufとHTTP/2により一般的に良い)	Protobuf
GraphQL	HTTP/1.1、HTTP/2	/graphqlの単一。一つのエンドポイントで全てのクエリとミューテーションを処理し、クライアントは具体的なデータ要求をクエリとして送信する。	◯(クライアントがほしいデータを指定するので、オーバ/アンダーフェッチは起こりにくい)	型付けられたGraphQLのスキーマ

引用: REST vs gRPC vs GraphQLの自分なりのまとめと結論

あんまり詳しくないですが、色々違うんだなということは分かりました。

作るもの

さてさて、なにはともあれ実装していきます。 そんなに難しいものを作るつもりはないのですが、リクエストに対して"hello world"を返すだけだとちょっとシンプル過ぎるので、少しだけ違うものを作ってみようと思います。

今回は「requestで2つのベクトルの名前を受取り、そのコサイン類似度をresponseする」サーバーについて考えてみようと思います。 概念図としてはこんな感じです。

レイヤー構成

今回は「いろんなサーバーを立てる」ことがやりたいことなので、こんな感じにしてアプリケーション層だけ切り替えるだけにします。 その他のロジックについては共通化して実装しようと思います。

イメージとしてはこんな感じです。

なのでビジネスロジック層とデータアクセス層は共通のものを使用しようと思います。

共通化部分を作る

最初に、本題とはあまり関係ないビジネスロジック層とデータアクセス層を作っちゃいます。 「requestで2つのベクトルの名前を受取り、そのコサイン類似度をresponseする」なので、ビジネスロジック層とデータアクセス層の役割はそれぞれこんな感じにします。

• ビジネスロジック層: 2つのベクトルを受け取り、コサイン類似度を計算して返す
• データアクセス層: ベクトルの名前から対応するベクトルを返す

データアクセス層については、ちゃんと作るならばベクトルをDBなどに入れてそこへのアクセスを書くのが筋でしょうが、今回はめんどくさい簡単のためjsonファイルで持っとくことにします。

import numpy as np
import json


class Vectors:
    def __init__(self):
        with open('cosine_similarity/vectors.json') as f:
            self.vector_dict = json.loads(f.read())

    def get_vector(self, name: str) -> np.ndarray:
        return np.array(self.vector_dict[name])

ビジネスロジック層について書いてみるとこんな感じです。

import numpy as np
from .data_access import Vectors


class Similarity:
    def __init__(self):
        self.vectores = Vectors()

    def calculate_similarity(self, name1: str, name2: str) -> float:
        vector1 = self.vectores.get_vector(name1)
        vector2 = self.vectores.get_vector(name2)

        if vector1.shape != vector2.shape:
            raise ValueError("Vectors must have the same shape")

        if vector1.ndim != 1 or vector2.ndim != 1:
            raise ValueError("Vectors must be 1-dimensional")

        if vector1 is None or vector2 is None:
            raise ValueError("Vectors cannot be None")

        return np.dot(vector1, vector2) / (np.linalg.norm(vector1) * np.linalg.norm(vector2))

これで、プレゼンテーション層はSimilarity.calculate_similarityメソッドを呼び出して、その応答を返せば良さそうです。

プレゼンテーション層

さて、本題のプレゼンテーション層を見ていきます。

REST

今回はAPを実装したいので、RESTについてはFastAPIを使用して実装します。

記述はめっちゃ簡単でこんな感じにするだけです。

from fastapi import FastAPI
from cosine_similarity.application import Similarity

app = FastAPI()
sim = Similarity()


@app.get("/")
async def root(v1: str, v2: str):
    return {"cosine_sim": sim.calculate_similarity(v1, v2)}

実際に動かしてみるとこんな感じです。

root@fe969dd02c0a:/src# uvicorn server:app --reload &
[1] 7
root@fe969dd02c0a:/src# INFO:     Will watch for changes in these directories: ['/src']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [7] using WatchFiles
INFO:     Started server process [9]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

root@fe969dd02c0a:/src# curl "http://127.0.0.1:8000/?v1=000&v2=010"
INFO:     127.0.0.1:57428 - "GET /?v1=000&v2=010 HTTP/1.1" 200 OK
{"cosine_sim":0.7455707409641453}

なんかそれっぽいcosine_simが返ってきているのでOKそうですね。

gRPC

gRPCはRequestとResponseのスキーマを定義してから使用します。

syntax = "proto3";

package cosine_sim;

service CosineSimService {
  rpc calc_cosine_sim (CosineSimRequest) returns (CosineSimResponse);
}

message CosineSimRequest {
    string v1 = 1;
    string v2 = 2;
}

message CosineSimResponse {
    float cosinesim = 1;
}

このprotoファイルをコンパイルして使用します。

python -m grpc_tools.protoc -I./proto --python_out=./pb --pyi_out=./pb --grpc_python_out=./pb ./proto/cosine_similarity.proto

これを使ってgRPCサーバーをこんな感じに書きます。

from concurrent import futures
import grpc
from pb import cosine_similarity_pb2
from pb import cosine_similarity_pb2_grpc
from cosine_similarity.application import Similarity


class CosineSim(cosine_similarity_pb2_grpc.CosineSimServiceServicer):

    def __init__(self) -> None:
        super().__init__()
        self.sim = Similarity()

    def calc_cosine_sim(self, request, context):
        v1 = request.v1
        v2 = request.v2
        return cosine_similarity_pb2.CosineSimResponse(
            cosinesim=self.sim.calculate_similarity(v1, v2)
        )


def serve():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    cosine_sim = CosineSim()
    cosine_similarity_pb2_grpc.add_CosineSimServiceServicer_to_server(
        cosine_sim, server
    )

    # サーバーの立ち上げ
    server.add_insecure_port('[::]:5001')
    server.start()
    print("server started")
    server.wait_for_termination()


if __name__ == '__main__':
    serve()

実際にリクエストしてみるとこんな感じです。

root@76601bd3576d:/src# python server.py &
[1] 36
root@76601bd3576d:/src# server started
root@76601bd3576d:/src# python client.py 
Response: 0.7590885162353516

こっちもそれっぽいコサイン類似度が計算されていることがわかります。

GraphQL

GraphQLについてはstrawberryとFastAPIを使用して実装してみます。

import strawberry
from fastapi import FastAPI
from strawberry.asgi import GraphQL
from cosine_similarity.application import Similarity


sim = Similarity()

@strawberry.type
class CosineSim:
    value: float

@strawberry.type
class Query:
    @strawberry.field
    def cosinesim(self, v1: int, v2: int) -> CosineSim:
        return CosineSim(value=sim.calculate_similarity(
                str(v1).zfill(3),
                str(v2).zfill(3)
            ))

schema = strawberry.Schema(query=Query)

graphql_app = GraphQL(schema)

app = FastAPI()
app.add_route("/graphql", graphql_app)
app.add_websocket_route("/graphql", graphql_app)

サーバーを立てて動作確認してみます。

root@917a51f31e5a:/src# uvicorn server:app --reload &
[1] 7
root@917a51f31e5a:/src# INFO:     Will watch for changes in these directories: ['/src']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [7] using WatchFiles
INFO:     Started server process [9]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

root@917a51f31e5a:/src# python client.py 
INFO:     127.0.0.1:43020 - "POST /graphql HTTP/1.1" 200 OK
{
  "cosinesim": {
    "value": 0.7424583353132823
  }
}

ちゃんと定義したcosine類似度が応答されました。ちゃんと動いてそうですね。

使用したコード

一応書いたコードはこちらに置いておきました。

github.com

参考文献

この記事を書くにあたって下記の文献を参考にさせていただきました。

• REST vs gRPC vs GraphQLの自分なりのまとめと結論
• MVC、3 層アーキテクチャから設計を学び始めるための基礎知識 #初心者 - Qiita
• 最初のステップ - FastAPI
• 30分でFastAPIでGraphQL APIを開発するチュートリアル
• pythonでgRPC通信をやってみる

感想

サーバーを立てる機会があったので、いっそのこといろんなサーバーの作り方の勉強がてら書いてみた記事でした。 あんまり普段気にしない世界だったので勉強になりました。