Safie Engineers' Blog!

Safieのエンジニアが書くブログです

FastAPI で Pydantic v2を使うと性能向上！

サーバー・インフラ

サーバーサイドエンジニアの松木 (@tatsuma_matsuki) です。

セーフィーではいくつかのサービスでFastAPIを使った開発を行っています。FastAPIでは、Pydanticというライブラリを使ってリクエスト・レスポンスのモデルのバリデーションなどを実装することができます。このPydanticというライブラリですが、近いうちにメジャーバージョンアップのリリースが予定されており、これによりモデルのバリデーション処理が高速化されることがアナウンスされています！

以下のページによると、Pydantic v2はv1に比べて4倍～50倍の性能向上があると書かれているので、これは期待してしまいます。

https://docs.pydantic.dev/latest/blog/pydantic-v2/#performance

そして、Pydanticを利用するFastAPIでも2023/6/21に 0.100.0-beta1 がリリースされ、このリリースでは、Pydantic v2を利用することができるようになっています！この記事の執筆時点では、Pydantic・FastAPIともにまだβ版ではありますが、試しにFastAPI + Pydantic v1 と FastAPI + Pydantic v2の性能を比較してみましたので、その内容をこの記事で共有したいと思います。

Pydantic BaseModel

PydanticのBaseModelについて簡単に説明します。BaseModelを継承したクラスでデータのモデルを定義することで、クラス変数に定義した型ヒントとインスタンス作成時に指定された値の型を比較し、正しい型なのかどうかをバリデーションしてくれます。型ヒント以外にも、独自のバリデーションルールを定義することもできます。以下がPydantic BaseModelを利用したバリデーションの例です。

from pydantic import BaseModel


class Service(BaseModel):
    service_id: int
    name: str

    # 独自のバリデーションルール
    @validator("name")
    def name_has_service(cls, v):
        if "Service" not in v:
            raise ValueError("name must have 'Service'")
        return v

Service(service_id=1, name="Service A")
# > OK

Service(service_id="a", name="Service A")
# > ValidationError

Service(service_id=1, name="A")
# > ValidationError

FastAPIを使っていると、必然的にこのPydantic BaseModelを使うことになるので、Pydantic は FastAPIで使うもの、というイメージを持たれるかもしれないですが、（私自身初めはそうでした）Pydantic の BaseModelは、FastAPI以外でも使うと結構便利です。

バリデーション以外にも、標準出力やdict型と相互変換、JSONシリアライズなどが簡単にできる他、インスタンス同士の比較もサクッとできてしまいます（単体テストなどを書く際に便利）。

from pydantic import BaseModel


class Service(BaseModel):
    service_id: int
    name: str


service_1 = Service(service_id=1, name="Service A")

print(service_1)
# > service_id=1 name='Service A'

print(service_1.dict())
# > {'service_id': 1, 'name': 'Service A'}

print(service_1.json())
# > {"service_id": 1, "name": "Service A"}

service_2 = Service(service_id=1, name="Service A")

print(service_1 == service_2)
# > True

service_3 = Service.parse_obj({"service_id": 1, "name": "Service A"})

print(service_1 == service_3)
# > True

Pydantic v2でモデルのバリデーションが高速化

Pydantic v2では、バリデーション部分の実装をRustで書き直し、pydantic-coreという別のパッケージに分離されています。これにより、v1と比較して、4～50倍バリデーション処理が高速化されたとPydantic公式ページで言及されています。

この記事では、このPydantic v1 → v2の性能向上がFastAPIで実装したAPIの応答時間にどの程度影響を与えるのかを調べるために、簡単なアプリケーション（APIサーバー）を実装して性能を測ってみます。

環境セットアップ

実際に簡単なAPIを実装して性能を比較するために、まずは環境のセットアップです。Python 3.10.6の環境で、まずはfastapiの最新バージョンをインストールします。

$ poetry add fastapi pydantic
Using version ^0.98.0 for fastapi
Using version ^1.10.9 for pydantic

Updating dependencies
Resolving dependencies... (0.2s)

Package operations: 8 installs, 0 updates, 0 removals

  • Installing exceptiongroup (1.1.1)
  • Installing idna (3.4)
  • Installing sniffio (1.3.0)
  • Installing anyio (3.7.0)
  • Installing typing-extensions (4.6.3)
  • Installing pydantic (1.10.9)
  • Installing starlette (0.27.0)
  • Installing fastapi (0.98.0)

Writing lock file

続いて、別の環境でpre-releaseのfastapiおよびpydanticをインストールします。pydanticが2.0b3、fastapiが0.100.0b1となっています。また、pydantic-coreというパッケージが増えていることも分かります。

$ poetry add fastapi pydantic --allow-prereleases
Using version ^0.100.0b1 for fastapi
Using version ^2.0b3 for pydantic

Updating dependencies
Resolving dependencies... (1.3s)

Package operations: 10 installs, 0 updates, 0 removals

  • Installing exceptiongroup (1.1.1)
  • Installing idna (3.4)
  • Installing sniffio (1.3.0)
  • Installing typing-extensions (4.6.3)
  • Installing annotated-types (0.5.0)
  • Installing anyio (3.7.0)
  • Installing pydantic-core (0.39.0)
  • Installing pydantic (2.0b3)
  • Installing starlette (0.27.0)
  • Installing fastapi (0.100.0b1)

Writing lock file

アプリケーションの実装

今回は単純にあるオブジェクトのリストを返すようなAPIをアプリケーションとして簡単に実装します。今回Pydantic v2で性能向上が期待できるのはBaseModelによる変数のバリデーションの部分ですので、複雑な情報を持ったオブジェクトのリストを返すようなAPIが一番効果が大きいのではないかと予想をしています。

以下のような Service というモデルのリストを返すAPIを実装しました。conintを使ったり、datetimeを使ったりして少し複雑にしてみましたが、比較的シンプルな仕様のAPIかと思います。

Pydantic v1用の実装は以下の通りです。

from datetime import datetime
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field, conint

app = FastAPI()


class Service(BaseModel):
    service_id: int = Field(..., title="Service ID")
    name: str = Field(..., title="Name of the service")
    version_number: conint(ge=1) = Field(..., title="Version of the service")
    version_name: str = Field(..., title="Version name of the service")
    created_on: datetime = Field(..., title="Date of the service was created")
    updated_on: datetime = Field(..., title="Date of the service was updated")


class ServiceList(BaseModel):
    offset: int = Field(..., title="Offset of the list")
    count: int = Field(..., title="Length of the list")
    total: int = Field(..., title="Total count of the available services")
    services: list[Service] = Field(..., title="List of the services")


@app.get("/services")
async def get_services(
    offset: int = Query(default=0, ge=0, title="Offset of the list"),
) -> ServiceList:
    services = []
    for i in range(1, 20):
        services.append(
            {
                "service_id": i,
                "name": f"service-{i}",
                "version_number": i,
                "version_name": f"v{i}",
                "created_on": datetime.now(),
                "updated_on": datetime.now(),
            }
        )

    return ServiceList(
        offset=offset,
        count=len(services),
        total=len(services),
        services=[Service.parse_obj(s) for s in services],
    )

ただサーバー側でリストを適当に作って ServiceList のモデルとして返しているだけですので、負荷としてはほとんどがモデルのバリデーション部分だけなのではないかと思っています。

次のコードがPydantic v2用のコードです。ほとんど同じですが、Pydantic v2では、BaseModelのparse_obj() のメソッドの名前が変わっているため、そこだけ少し修正しています。

from datetime import datetime
from fastapi import FastAPI, Query
from pydantic import BaseModel, Field, conint

app = FastAPI()


class Service(BaseModel):
    service_id: int = Field(..., title="Service ID")
    name: str = Field(..., title="Name of the service")
    version_number: conint(ge=1) = Field(..., title="Version of the service")
    version_name: str = Field(..., title="Version name of the service")
    created_on: datetime = Field(..., title="Date of the service was created")
    updated_on: datetime = Field(..., title="Date of the service was updated")


class ServiceList(BaseModel):
    offset: int = Field(..., title="Offset of the list")
    count: int = Field(..., title="Length of the list")
    total: int = Field(..., title="Total count of the available services")
    services: list[Service] = Field(..., title="List of the services")


@app.get("/services")
async def get_services(
    offset: int = Query(default=0, ge=0, title="Offset of the list"),
) -> ServiceList:
    services = []
    for i in range(1, 21):
        services.append(
            {
                "service_id": i,
                "name": f"service-{i}",
                "version_number": i,
                "version_name": f"v{i}",
                "created_on": datetime.now(),
                "updated_on": datetime.now(),
            }
        )

    return ServiceList(
        offset=offset,
        count=len(services),
        total=len(services),
        services=[Service.model_validate(s) for s in services],  # parse_obj()は使えない
    )

Pydantic v1 → v2はかなり非互換のある変更が含まれており、移行には少なからずコードへの修正も合わせて必要になると思います。PydanticのページにMigration Guideが用意されていますので、詳しくは以下のページを参照ください。

https://docs.pydantic.dev/dev-v2/migration/

Locustで性能測定

さっそく、上で実装した簡易的なアプリケーションをLocustを使って性能測定したいと思います。

Locustのコードは以下のように書きました。（ChatGPTでほぼ同じコードが一瞬で生成できます）

# main.py
from locust import HttpUser, task, constant_throughput


class ServicesList(HttpUser):
    host = "http://localhost:8001"
    wait_time = constant_throughput(1)

    @task(1)
    def get_services(self):
        with self.client.get("/services", catch_response=True) as response:
            if response.status_code != 200:

以下のようにPydantic v1, v2を使ったアプリケーションをそれぞれ起動します。

$ poetry add uvicorn[standard]
...
$ poetry run uvicorn main:app --host 0.0.0.0 --reload --port 8000
INFO:     Will watch for changes in these directories: ['/tmp/fastapi-0.100.0-beta1']
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [14005] using WatchFiles
INFO:     Started server process [14050]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

Locustのコードは、以下のように実行します。引数でユーザー数を指定することで、リクエスト数をコントロールします。

$ poetry run locust -f main.py --headless --users 10 --spawn-rate 1 --csv=outputs/fastapi_pydantic_v1 -t 1m

...

Response time percentiles (approximated)
Type     Name                                                                                  50%    66%    75%    80%    90%    95%    98%    99%  99.9% 99.99%   100% # reqs
--------|--------------------------------------------------------------------------------|--------|------|------|------|------|------|------|------|------|------|------|------
GET      /services                                                                              13     16     18     20     27     31     38     41     45     45     45    555
--------|--------------------------------------------------------------------------------|--------|------|------|------|------|------|------|------|------|------|------|------

性能測定の結果

測定結果です！今回は、FastAPIのサーバーの起動は、ローカルのLaptop PC上で行っています。

以下が測定結果をグラフ化した結果です。横軸はLocust実行時に指定した設定したユーザー数で、縦軸が応答時間（ミリ秒）です。得られた結果を各統計値（中央値、90パーセンタイル値、95パーセンタイル値、99パーセンタイル値）毎にラインにして描いてみました。

性能測定結果比較

結果は一目瞭然ですね！ざっくり言うと、2～3倍高速化されていると思います！モデルのバリデーション部分が4～50倍の性能向上があるという話でしたので、今回のようなバリエーション処理以外ほぼ何もない実装だともっと劇的に向上するかも？と思っていましたが、思ったほどではなかったです。

もちろん、今回の結果はあくまで一例であり、実際に商用運用されているアプリケーションではどうなるかやってみないと分かりませんが、見積もりとして2倍以上の改善が見込めるのであれば、多少苦労してでも、Pydantic v2に移行する価値はありそうです！

まとめ

今回は、簡単なAPIを実装して応答時間を測定してみましたが、実サービスのAPIでは、データベース等へのアクセスが挟まるので、実際にはその部分が支配的になり、応答時間に対してはあまり効果が出ないかも？とは思っていますが、FastAPIは基本そのようなI/Oを非同期で処理するので、APIサーバーでの負荷は結構下がるのかもしれません。早く実サービスでも試してみたいです！

Pydantic v1 → v2へのバージョンアップは非互換のある変更を多く含んでいますので、今後慎重にバージョンアップしていきたいと思います！

© Safie Inc.