Pydanticを使った型安全なデータ構造を持ったPythonコードのススメ

最終更新日 2023年8月27日

実行時の環境

Python 3.11.2

Pydantic 2.3.0

FastAPI 0.103.0

SQLAlchemy 2.0.20

uvicorn 0.23.2

BtoC向けサービスのコードを書いてテストを実施し、デプロイしたら予期しないエラーが発生した、ということはエンジニアの誰しもが経験したことがあるでしょう。
テストがすべての事故を防いでくれるわけではありません。エンドユーザーは開発者が予想もしない入力をすることがあり、通信されるデータの形式のすべてを把握することは非常に難しいです。

こういったバグを出してしまい、プロダクトの安全性を担保できないことはエンジニアにとって許容し難いものです。
Pydanticを使えば、この問題に対処する堅牢性アップへ貢献できます。

Pyndanticとは

2023年8月現在、世界で最も利用されているPythonのデータバリデーションライブラリがPydanticです。
現在はバージョン2.3.0になり、アップデートが早いためライブラリの有用性も短期サイクルで向上しています。

多くのlinterやIDEにも対応しているため、Pythonエンジニアは採用するべきライブラリです。

Pydanticの特徴

強力なタイプヒンティング
コア部分がRust製なため高速
JSONスキーマ生成によるOpenAPIフレンドリーさ
バリデーションモードを「厳格」「緩め」で選べる
dataclassやTypeDictといった組み込みライブラリとの親和性
バリデーターの高いカスタマイズ性
FastAPIやhuggingfaceなど多くのライブラリがPydanticを内包
FAANGなど有名企業が採用するライブラリ

軽く触ってみる

ドキュメントを読むより触ったほうが早いので触ってみましょう。
pipでインストールできます。

pip install pydantic

サンプルとして、注文情報のモデルを作成してみます。
モデルは、エンティティ（データ）に対する種類や内容を定義すること、と覚えておけばOKです。

from datetime import datetime as dt
from pydantic import BaseModel, PositiveInt


class Order(BaseModel):
    id: int
    customer_name: str = 'ヤマダタロウ'
    ordered_at: dt | None = dt.now()
    price: PositiveInt = 1000


order_data = {'id': 111,
              'customer_name': 'ヤマダハナコ',
              'ordered_at': '2023-08-27 12:00:00',
              'price': 2000}

order = Order(**order_data)

print(order.model_dump())

これを実行した結果は以下です。

$ python app.py
{'id': 111, 'customer_name': 'ヤマダハナコ', 'ordered_at': datetime.datetime(2023, 8, 27, 12, 0), 'price': 2000}

これは成功例ですが、このコードのid部分を変更して文字列にして再実行みましょう。

order_data = {'id': 'testID',
              'customer_name': 'ヤマダハナコ',
              'ordered_at': '2023-08-27 12:00:00',
              'price': 2000}

これを実行すると以下のようなエラーが出ます。

pydantic_core._pydantic_core.ValidationError: 1 validation error for Order

このように、定義された型に対して異なる型が入力されるとバリデーションエラーを出し、データ保全を助けてくれるのがPydanticです。

FastAPI + Pydanticでの使用例

Pydanticを利用する取っ掛かりとして最適なFastAPIアプリケーションの実装例を見ていきましょう。
FastAPIはPythonのミニマルなRESTful APIアプリケーションフレームワークです。しばしばフロントエンドフレームワークと組み合わせて使われます。

今回は例として、商品テーブルを参照するRESTful APIを作る前提で、Product（商品）モデルを使うアプリケーションを作っていきます。

準備として、fastapi、uvicorn、pydantic、sqlalchemyをインストールしましょう。

FastAPIのデータベース接続など、冗長な実装は本題のPydanticから外れるため省いています。

pip install fastapi uvicorn pydantic pytest sqlalchemy

fastapi: REST API用軽量フレームワーク
uvicorn: ASGI Webサーバー
pydantic: バリデーション
sqlalchemy: ORMライブラリ

ディレクトリ構成

最終的なディレクトリ構成は以下となります。

├── app.py # エントリーポイント
├── models # モデル
│   ├── base.py
│   └── product.py
├── schemas # リクエスト、レスポンスなどの定義
│   ├── __init__.py
│   └── product.py
└── services # ルーターと共にロジックを提供
    └── product.py

モデルの定義

models/ディレクトリ配下にbase.pyを作りましょう。
このbase.pyはベースモデルを定義します。今回はただsqlalchemyのベース宣言クラスのDeclarativeBaseを継承するだけです。通常は、ここでidやcreated_atなどの共通カラムの定義やデータベース接続を記載します。

from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

次にmodels/product.pyを見てみましょう。

from sqlalchemy import Integer, String
from sqlalchemy.orm import Mapped, mapped_column

from pydantic import PositiveInt

from .base import Base


class Product(Base):
    __tablename__ = "products"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(255), nullable=False)
    price: Mapped[PositiveInt] = mapped_column(Integer, nullable=False)
    description: Mapped[str] = mapped_column(String(255), nullable=True)

ここではproductsテーブルに「ID」「商品名」「価格」「商品説明」を定義しています。先ほど作ったbase.pyのBaseを継承しています。PydanticにBaseModelが用意されていますが、sqlalchemyと組み合わせるためここでは使用しません。

リクエストとレスポンスを定義

モデルの定義とは別に、HTTPリクエストされた際に「どのような値を受け取るか」「どのような値を返すか」を定義します。
同じファイル名ですが、次はschemas/ディレクトリ以下のschemas/product.pyを以下のようにします。

from pydantic import BaseModel, ConfigDict


class ProductBase(BaseModel):
    name: str
    price: int
    description: str = None


class ProductCreate(ProductBase):
    pass


class ProductResponse(ProductBase):
    id: int

    # 今回は省略しているが、from_attributesはmodel_validateを使う設定
    model_config = ConfigDict(from_attributes=True)

ProductBaseで共通となるフィールドを定義しています。
schemasをロードしたら使えるようにするため、__init__.pyも以下のようにします。

from .product import ProductCreate, ProductResponse

HTTPリクエストを受け取る

上記で定義した内容を受け取るアプリケーションを作りましょう。
services/product.pyを以下のようにします。

from fastapi import APIRouter

import schemas
from models.product import Product

router = APIRouter()


@router.get("/{id}", operation_id="get_product")
async def get_product(id: int) -> schemas.ProductResponse:
    return Product(
        id=id,
        name=f"Product {id}",
        price=100,
        description=f"Description of product {id}",
    )


@router.post("", operation_id="create_product")
async def create_product(product: schemas.ProductCreate) -> schemas.ProductResponse:
    return Product(
        id=2,
        name=product.name,
        price=product.price,
        description=product.description,
    )

FastAPIのルーティング機能を使います。これを利用するためにapp.pyを以下のように書きます。

from fastapi import FastAPI

from services import product

app = FastAPI()

app.include_router(product.router, prefix="/product", tags=["product"])

このようにすることで[GET]/product/{id}と[POST]/productにルーターを設定できます。

以下のコマンドでアプリケーションを立ち上げましょう。

uvicorn app:app --reload --host 0.0.0.0 --port 8080

http://localhost/product/1にブラウザでアクセスすると、以下のようなJSONが返ってきます。

{
  "name": "Product 1",
  "price": 100,
  "description": "Description of product 1",
  "id": 1
}

さらにcurlコマンドを使ってcreateのAPIにもアクセスしてみましょう。

$ curl -X POST -H 'Content-type: application/json' --data '{"name":"Product2", "price": 200, "description": "Description of product 2"}' http://localhost:8080/product

これで以下のようなレスポンスが返ってきたら完了です。

{
  "name": "Product2",
  "price": 200,
  "description": "Description of product 2",
  "id": 2
}

エラーケースも試しておきましょう。本来であればcreateのAPIにはnameフィールドは必須ですが、これを省いてリクエストしてみます。

$ curl -X POST -H 'Content-type: application/json' --data '{"price": 200, "description": "Description of product 2"}' http://localhost:8080/product

このリクエストに対する結果は以下です。

{
  "detail": [
    {
      "type": "missing",
      "loc": [
        "body",
        "name"
      ],
      "msg": "Field required",
      "input": {
        "price": 200,
        "description": "Description of product 2"
      },
      "url": "https://errors.pydantic.dev/2.3/v/missing"
    }
  ]
}

正しくエラーとして出力されています。

Pydanticの便利機能

実装例を紹介すると長くなりすぎるため、ここでは省略しますが、Pydanticには以下のような機能があります。

computed_field

computed_fieldは、@propertyや@cached_propertyデコレーターに関数の返り値を付与できます。
以下は商品の単体重量に対して数量をかけた総重量を付与した例です。

from pydantic import BaseModel, computed_field


class Product(BaseModel):
    unit_weight: int
    quantity: int

    @computed_field
    @property
    def total_weight(self) -> int:
        return self.unit_weight * self.quantity

print(Product(unit_weight=10, quantity=5).model_dump())
# => {'unit_weight': 10, 'quantity': 5, 'total_weight': 50}

バリデーションデコレーター

個人的に便利だなぁと感じるバリデーションデコレーターである@validate_callの紹介です。
関数の呼び出しの前に引数チェックが行なえます。

from pydantic import ValidationError, validate_call


@validate_call
def say_hello(name: str, age: int = 17) -> str:
    return f"Hello, {name}! You are {age} years old."

print(say_hello("Tanabe", 34))
print(say_hello("Hanako"))

try:
    print(say_hello("Kei", "Suke"))
except ValidationError as e:
    print(e)

これの出力結果は以下です。

Hello, Tanabe! You are 34 years old.
Hello, Hanako! You are 17 years old.
1 validation error for say_hello
1
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='Suke', input_type=str]

実際に関数の呼び出しが行われることはないため、より安全な実装に役立ちます。

Strict Modeでより厳格にする

Pydanticはデフォルトでlax mode（緩めモード）という状態です。言葉が意味する通り、整数値を文字列として渡しても自動で解釈してくれます。
これは助かるケースもありますが、より堅牢な実装をしたいときは以下のようにstrict modeで実装するべきです。

from pydantic import ValidationError, BaseModel


class Product(BaseModel):
    price: int

print(Product.model_validate({"price": '100'}))

try:
    Product.model_validate({"price": '100'}, strict=True)
except ValidationError as e:
    print(e)

上記の出力結果は以下です。

price=100
1 validation error for Product
price
  Input should be a valid integer [type=int_type, input_value='100', input_type=str]

Pydanticの設定ファイル強化

Pydanticの設定ファイルの拡張にpydantic-settingsがあり、便利な設定値として定義できます。
設定値のプリフィックスや、dotenvのサポート、そしてDockerのSecrets対応など、本番環境で使いやすい機能が揃っています。

pipでインストールしましょう。

pip install pydantic-settings

自身で設定したい項目をまとめて設定可能なので、保守性にも優れています。

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict
import os


class MySQLSettings(BaseSettings):
    host: str
    port: int
    user: str
    password: str
    database: str


class Settings(BaseSettings, case_sensitive=True):
    api_key: str = Field('xxx', alias='app_api_key')
    mysql: MySQLSettings


os.environ['mysql'] = '{"host": "localhost", "port": 3306, "user": "testuser", "password": "p@ssw0rd", "database": "testdb"}'

print(Settings().model_dump())

この出力結果は以下です。

{
  "api_key": "xxx",
  "mysql": {
    "host": "localhost",
    "port": 3306,
    "user": "testuser",
    "password": "p@ssw0rd",
    "database": "testdb"
  }
}

エラーハンドリング

エラーハンドリングは基本的にValidationErrorに対するraiseで事足ります。
キャッチした例外がとてもわかりやすいので、エラー発生箇所が特定しやすくなっています。

try:
    # Hoge
except ValidationError as e:
    print(e)

# e.errors = 見つかったエラー一覧
# e.error_count = エラーの数
# e.json = JSON形式

ロバストなPythonコーディング

今回はPydanticの基本機能を紹介しましたが、OpenAPIでAPI仕様を確認したり、単体テストを書いたりなど、本番では考慮する点がまだまだあります。
堅牢性の高いアプリケーションの構築は、すべてのPythonアプリケーションに対して毎回存在する課題です。

エンジニアとして、質の高いコーディングをしたいのはもちろんですが、Pydanticのようなライブラリを効率的に利用することで省コード化していきましょう。意図しない入力を防ぐことができれば、それだけで大幅に事故防止へつながります。