当サイトは、アフィリエイト広告を利用しています

【Python × Flask】pydanticでREST APIのバリデーションを行う方法

作成日：2024月08月11日
更新日：2024年08月21日

PythonのFlask製のREST APIでリクエストとレスポンスのバリデーションを
行う方法をまとめる。

Flask自体には組み込みでバリデーション機能ないので
外部ライブラリの「pydantic」を使う。

pydanticは「2.8.2」を使う。
※v2.xから書き方の変更が割とある。

pydanticとは？

pydanticは型ヒントを使用してデータのバリデーションを行うライブラリ。

pydanticの特徴

pydanticは

  • データモデルの定義
  • データバリデーション
  • データ変換
  • データのシリアライズ

を行うことができる。

サンプルプログラムで試してみる。

データモデルの定義

pydanticのBaseModelクラスを継承した
データモデルを定義する。

データモデル
from pydantic import BaseModel


class User(BaseModel):
    user_id: int
    name: str
    age: int

データモデルは

  • 自動バリデーション
  • データの型変換

に使用する。

データバリデーション

データモデルの型ヒントを使用してバリデーションを行う。

バリデーション
from pydantic import BaseModel, ValidationError


class User(BaseModel):
    user_id: int
    name: str
    age: int


try:
    # `user_id`を指定せずにインスタンスを作成する場合、バリデーションエラーが発生する
    user = User(name="Alice", age=25)
    print(user)
except ValidationError as e:
    print(e.json())


# 実行結果
# [{"type":"missing","loc":["user_id"],"msg":"Field required","input":{"name":"Alice","age":25},"url":"https://errors.pydantic.dev/2.8/v/missing"}]

user_idがないためバリデーションエラーとなる。

データの型変換

バリデーションがすべて成功した場合、Pydanticは渡されたデータを使ってUserクラスのインスタンス（オブジェクト）
を作成する。
このインスタンスは、各フィールドがバリデート済みのデータを持つPythonのデータクラスとして扱うことができる。

データ変換
from pydantic import BaseModel, ValidationError


class User(BaseModel):
    user_id: int
    name: str
    age: int


try:
    user = User(user_id=2, name="Alice", age=25)
    print(user)


    # インスタンスが`User`のものであることを確認
    if isinstance(user, User):
        print("`user`はUserのインスタンスです")
    else:
        print("`user`はUserのインスタンスではありません")
except ValidationError as e:
    print(e.json())


# 実行結果
# user_id=2 name='Alice' age=25
# `user`はUserのインスタンスです

バリデーションとデータ変換（デシリアライズ）は同時に行われる。

データのシリアライズ

Userクラスのインスタンスからシリアライズする。

シリアライズ
from pydantic import BaseModel
import json  # 標準ライブラリのjsonモジュールをインポート


class User(BaseModel):
    user_id: int
    name: str
    age: int


user = User(user_id=1, name="Charlie", age=40)


# 辞書形式に変換（model_dump()を使用）
user_dict = user.model_dump()
print(user_dict)


# JSON形式に変換（json.dumps()を使用）
user_json = json.dumps(user_dict)
print(user_json)


# 実行結果
# {'user_id': 1, 'name': 'Charlie', 'age': 40}
# {"user_id": 1, "name": "Charlie", "age": 40}

REST APIの場合はレスポンスはJSONにしてから返却するので
その時に使用する。

FlaskのREST APIでpydanticのバリデーションを使う

FlaskのREST APIでpydanticを使って

  • リクエストのバリデーション
  • レスポンスのバリデーション

の機能を実装してみる。

環境

開発環境としては、dockerを使ったコンテナ環境をつくり
動作を確認するので環境条件は下記のようにする

  • Windows10
  • Docker version 24.0.2(Docker for Windows)
  • Flask 3.0.2
  • pydantic 2.8.2

構成

プロジェクトの構成は下記にする

bash
.
|-- .env
|-- .vscode
|   |-- launch.json
|   `-- tasks.json
|-- Dockerfile
|-- api
|   |-- __init__.py
|   |-- api.py
|   |-- schemas.py
|   `-- wsgi.py
|-- docker-compose.yml
|-- requirements.txt
`-- vscode_ex_install.sh

flaskのREST APIを

  • dockerコンテナで開発環境を作る
  • サーバー起動
  • 動作確認
  • デバッグ方法

する方法については下記記事で紹介しているので割愛する

上記記事で作ったプロジェクトに対して

  • schemas.pyの追加
  • api.pyの変更
  • requirements.txtの変更※pydanticの追加

をしているので他のサーバー起動やデバッグはそのまま流用できる。

schemas.pyの追加

バリデーションを行うためのデータモデルを定義する

schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Optional


class Users(BaseModel):
    user_id: str
    name: str = Field(..., min_length=1, max_length=50)
    age: Optional[int] = Field(None, ge=0, le=120)


    @field_validator('name')
    def name_must_be_capitalized(cls, v):
        if not v.istitle():
            raise ValueError('名前は大文字で始める必要があります')
        return v

型ヒントでバリデーションを行うことができる

user_id

文字列型の必須フィールド

name

文字列型の必須フィールド。
またFieldを使って下記の制約をしている

  • 文字列の長さは1文字以上
  • 文字列の長さは50文字以下

age

整数型のフィードで必須ではない。 またFieldを使って下記の制約をしている

  • 値は0以上
  • 値は120以下
  • デフォルト値はNone※指定されない場合はNoneが入る

カスタムバリデーション

nameに対してカスタムバリデーションを設定。
nameフィールドの値がタイトルケース（つまり、各単語が大文字で始まる）でない場合に
バリデーションエラーを発生させる。

バリデーションを実装する

api.pyを変更してREST APIの

  • POSTリクエスト処理（登録）でリクエストボディのバリデーション
  • GETリクエスト処理（指定参照）でレスポンスのバリデーション

を行う。

api.py全体のソースを載せる。

api.py
from flask import Flask, request, jsonify,make_response
import copy


from pydantic import ValidationError
from api.schemas import Users


# ディクショナリ
users = [
    {"user_id": "1", "name": "Tujimura", "age": 11},
    {"user_id": "2", "name": "mori", "age": 20},
    {"user_id": "3", "name": "shimada", "age": 50},
    {"user_id": "4", "name": "kyogoku", "age": 70}]


# 関数ベースのルーティング関数
def routers(api):


    # バリデーションエラーハンドラー
    @api.errorhandler(ValidationError)
    def handle_validation_error(error):
        print(error.json())


        response = {
            "status_code": "422",
            "detail": str(error.errors())
        }
        # カスタム属性 `context` でリクエストかレスポンスのエラーかを区別
        if hasattr(error, 'context') and error.context == 'request':
            response["error_message"] = "Request validation error"
            return make_response(jsonify(response), 422)
        elif hasattr(error, 'context') and error.context == 'response':
            response["error_message"] = "Response validation error"
            return make_response(jsonify(response), 422)


    # user_idを指定して取得
    @api.route('/<user_id>', methods=['GET'])
    def get_user_by_id(user_id):
        user = next((user for user in users if user['user_id'] == user_id), None)


        # レスポンスバリデーション
        try:
            # バリデーション実行
            validated_data = Users(**user)
        except ValidationError as e:
            e.context = 'response'
            raise e        


        if user:
            return make_response(jsonify(validated_data.model_dump()),200)


        return make_response(jsonify({"error": "User not found"}), 404)


    # 登録
    @api.route('/', methods=['POST'])
    def post_user():
        # リクエストバリデーション
        try:
            # リクエストボディをdictに変換
            request_data = request.get_json()
            # バリデーション実行
            validated_data = Users(**request_data)
            # validated_data = Users.model_validate(request_data)
        except ValidationError as e:
            e.context = 'request'
            raise e


        res_users = copy.deepcopy(users)
        res_users.append(validated_data.model_dump())
        return make_response(jsonify(res_users), 201)


# flaskインスタンス作成
def create_app():
    # 作成
    api = Flask(__name__)


    # 日本語文字化け対応
    api.json.ensure_ascii = False


    # 関数ベースのルーティング設定
    routers(api)


    return api
  • ValidationErrorハンドラー関数を設定
  • POSTでリクエストのバリデーション処理を実装
  • GETでレスポンスのバリデーション処理を実装
    ※バリデーションしないエンドポイントは割愛しています。

細かく解説する。

ValidationErrorのハンドラー関数を設定

handle_validation_error
# バリデーションエラーハンドラー
    @api.errorhandler(ValidationError)
    def handle_validation_error(error):
        print(error.json())


        response = {
            "status_code": "422",
            "detail": str(error.errors())
        }
        # カスタム属性 `context` でリクエストかレスポンスのエラーかを区別
        if hasattr(error, 'context') and error.context == 'request':
            response["error_message"] = "Request validation error"
            return make_response(jsonify(response), 422)
        elif hasattr(error, 'context') and error.context == 'response':
            response["error_message"] = "Response validation error"
            return make_response(jsonify(response), 422)

handle_validation_errorをエラーハンドラー関数として登録することで
アプリケーション内で発生したValidationErrorはすべてこの関数で処理させる。
※try~exceptで捕まえた場合を除く。

handle_validation_errorは例外オブジェクトを引数として受け取り
コンテキストの値でリクエストとレスポンスのどちらのバリデーションエラーかを
判定して処理する。

POSTでリクエストのバリデーション処理を実装

リクエストバリデーション
# 登録
    @api.route('/', methods=['POST'])
    def post_user():
        # リクエストバリデーション
        try:
            # リクエストボディをdictに変換
            request_data = request.get_json()
            # バリデーション実行
            validated_data = Users(**request_data)
            # validated_data = Users.model_validate(request_data)
        except ValidationError as e:
            e.context = 'request'
            raise e


        res_users = copy.deepcopy(users)
        res_users.append(validated_data.model_dump())
        return make_response(jsonify(res_users), 201)

「validated_data = Users(**request_data)」でバリデーションを実行.
schemas.pyで定義したPydanticのBaseModelを継承したUserクラスは、受け取ったキーワード引数をもとに
自動的にバリデーションと型変換を行い、インスタンスを作成する。

ValidationErrorが発生した場合は例外オブジェクトにコンテキストとして
「request」を追加して、再スローする。

ValidationErrorだけではリクエストのバリデーションエラーか、レスポンスのバリデーションエラー
かを判定できないので、コンテキスト情報を追加して再スローしている。
※再スロー後はハンドラー関数で処理される

コメントアウトしている「Users.model_validate」を使っても
バリデーションを実行できるが、基本的には「Users(**request_data)」で良いと思う。

GETリクエストでレスポンスのバリデーション処理を実行

レスポンスのバリデーション
# user_idを指定して取得
    @api.route('/<user_id>', methods=['GET'])
    def get_user_by_id(user_id):
        user = next((user for user in users if user['user_id'] == user_id), None)


        # レスポンスバリデーション
        try:
            # バリデーション実行
            validated_data = Users(**user)
        except ValidationError as e:
            e.context = 'response'
            raise e        


        if user:
            return make_response(jsonify(validated_data.model_dump()),200)


        return make_response(jsonify({"error": "User not found"}), 404)

取得した値に対してバリデーションを実行する。
基本的にはリクエストのバリデーションと同じ。

ValidationError発生時はコンテキスト情報として、「response」を追加して
再フローする。
※再スロー後はハンドラー関数で処理される

動作確認

実装したバリデーションが機能するか確認する
サーバーの起動は下記で行う

bash
flask run --host=0.0.0.0

コンテナ上で動かす前提。

登録処理

登録時のリクエストバリデーションが作動することを確認する ホストからcurlで試す。

登録
$ curl -X POST http://localhost:5000 -H "Content-Type: application/json" -d '{  "user_id": "1",  "name": "tujimura",  "age": 11}'
{
  "detail": "[{'type': 'value_error', 'loc': ('name',), 'msg': 'Value error, 名前は大文字で始める必要があります', 'input': 'tujimura', 'ctx': {'error': ValueError('名前は大文字で始める必要があります')}, 'url': 'https://errors.pydantic.dev/2.8/v/value_error'}]",
  "error_message": "Request validation error",
  "status_code": "422"
}

カスタムバリデーションが機能している。

指定参照処理

指定参照
$ curl -X GET http://localhost:5000/2
{
  "detail": "[{'type': 'value_error', 'loc': ('name',), 'msg': 'Value error, 名前は大文字で始める必要があります', 'input': 'mori', 'ctx': {'error': ValueError('名前は大文字で始める必要があります')}, 'url': 'https://errors.pydantic.dev/2.8/v/value_error'}]",
  "error_message": "Response validation error",
  "status_code": "422"
}

2のデータは

2のデータ
{"user_id": "2", "name": "mori", "age": 20},

であるため、カスタムバリデーションでエラーになる。

1の場合は

1のデータ
{"user_id": "1", "name": "Tujimura", "age": 11},

なので

成功
$ curl -X GET http://localhost:5000/1
{
  "age": 11,
  "name": "Tujimura",
  "user_id": "1"
}

バリデーションエラーは発生しない

バリデーション関数を作成する

上記ではバリデーションを行うためにtry~exceptで囲んでいたが
エンドポイントが増えてくると同じことを繰り返し書く必要があり
無駄になるため、バリデーション関数の関数を作成して効率的にバリデーションを行ってみる

api.pyを変更する

api.py
from flask import Flask, request, jsonify,make_response
import copy


from pydantic import ValidationError, BaseModel
from typing import Optional, Type
from api.schemas import Users


# ディクショナリ
users = [
    {"user_id": "1", "name": "Tujimura", "age": 11},
    {"user_id": "2", "name": "mori", "age": 20},
    {"user_id": "3", "name": "shimada", "age": 50},
    {"user_id": "4", "name": "kyogoku", "age": 70}]


def validation_check(data: dict, model: Type[BaseModel],target: str):
    # レスポンスのバリデーションチェック
    try:
        validated_data = model(**data)  # 引数で渡されたモデルを使用してバリデーション
        return validated_data
    except ValidationError as e:
        if target == 'request':
            e.context = 'request'  # エラーオブジェクトにコンテキスト情報を追加
        else:
            e.context = 'response'  # エラーオブジェクトにコンテキスト情報を追加
        raise e # eを再スロー


# 関数ベースのルーティング関数
def routers(api):


    # バリデーションエラーハンドラー
    @api.errorhandler(ValidationError)
    def handle_validation_error(error):
        print(error.json())


        response = {
            "status_code": "422",
            "detail": str(error.errors())
        }
        # カスタム属性 `context` でリクエストかレスポンスのエラーかを区別
        if hasattr(error, 'context') and error.context == 'request':
            response["error_message"] = "Request validation error"
            return make_response(jsonify(response), 422)
        elif hasattr(error, 'context') and error.context == 'response':
            response["error_message"] = "Response validation error"
            return make_response(jsonify(response), 422)


    # user_idを指定して取得
    @api.route('/<user_id>', methods=['GET'])
    def get_user_by_id(user_id):
        user = next((user for user in users if user['user_id'] == user_id), None)


        # レスポンスバリデーション
        validated_data = validation_check(data=user, model=Users, target='response')


        if user:
            return make_response(jsonify(validated_data.model_dump()),200)


        return make_response(jsonify({"error": "User not found"}), 404)


    # 登録
    @api.route('/', methods=['POST'])
    def post_user():
        # リクエストバリデーション
        request_data = request.get_json()
        validated_data = validation_check(data=request_data, model=Users, target='request')


        res_users = copy.deepcopy(users)
        res_users.append(validated_data.model_dump())
        return make_response(jsonify(res_users), 201)


# flaskインスタンス作成
def create_app():
    # 作成
    api = Flask(__name__)


    # 日本語文字化け対応
    api.json.ensure_ascii = False


    # 関数ベースのルーティング設定
    routers(api)


    return api

validation_check関数内でバリデーションを行うようにする。

まとめ

flaskでpydanticを使ってリクエストとレスポンスのバリデーション機能を
実装してみた。

pydantic自体もシンプルで使いやすいのでFlaskでバリデーション機能を
実装する際の選択肢としてはかなり有力だと思う。

またFlaskのREST APIでバリデーションチェックにpydanticを使用している
実例として下記の記事が参考になるかと思います

新着記事

タグ別一覧
top