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

【FastApi × Poetry】pydanticでリクエスト・レスポンスをバリデーションする

作成日：2024月09月22日

更新日：2024年09月27日

PythonのFastApiで作ったREST APIでリクエストとレスポンスを
バリデーションライブラリであるpydanticを使って行ってみる。

タイトルではリクエスト・レスポンスのバリデーションと書いたが
その他にpydanticを使ってできることもまとめておく。
※バリデーションエラーのエラーハンドリングも

FastApiのレスポンス生成については下記でまとめている

動かして学ぶ！Python FastAPI開発入門（NEXT ONE） [ 中村翔 ]

pydanticとは？

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

pydantic

なぜFastApiでpydanticを使うのか？

FastApiとPydanticは親和性が非常に高いため。
というかFastApiがPydanticをネイティブに統合している。

FastApi自体がPydanticに依存しており、FastApiをインストールすると、
Pydanticも自動的にインストールされる。

つまり、FastApiを使う場合はPydanticはインストールなしの使うことができるのでバリデーションライブラリとして他のものを使う必要はない

またPydanticは型チェックやバリデーションの処理が非常に効率的で、
高速なパフォーマンスなのでリアルタイム性が求められるAPI開発でもパフォーマンスが損なわれない。

ちなみにFlaskではデフォルトでのバリデーション機能はないのでpydantic等の
ライブラリを使う必要があり、使ったとしてはFastApiほど簡単には実装できない。
※ネイティブに統合してるだけあり、FastApiの方が圧倒的に使いやすい

flaskでpydanticを使ってバリデーションする方法は一応下記記事で紹介している

pydanticでできること

pydanticを使うことで

リクエストバリデーション
レスポンスバリデーション
データシリアライゼーション・デシリアライゼーション
データシリアライゼーション・デシリアライゼーション

ができる。

実際にFastApiでREST APIを作って動作を確認する

リクエストバリデーション

リクエストバリデーションをpydanticを使ってやってみる。
pydanticによってリクエストバリデーションが実行されるタイミングは

エンドポイントに通信が来た時

になる。

main.py


@app.delete("/{user_id}")
async def delete_user(user_id: int):

    # 削除後のディクショナリを取得（実際に削除はしない）
    res_users = list(filter(lambda user: user['user_id'] != user_id, users))
    return JSONResponse(status_code=200, content=res_users)

であれば、バリデーションに成功した場合、エンドポイント関数のdelete_userの引数user_idに
バリデーション済みの値が渡される。

リクエストバリデーションエラー（例えば、リクエストボディやクエリパラメータの検証に失敗した場合）は
デフォルトで「422 Unprocessable Entity」を返す

リクエストは分けると

パスパラメータ
クエリパラメータ
リクエストボディ

になるので、それぞれについてバリデーションしてみる

パスパラメータ

パスパラメータに対するバリデーションを行う

main.py


from fastapi import FastAPI
from fastapi.responses import JSONResponse

app = FastAPI()

# ディクショナリ
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}
]

# user_id指定で削除
# パスパラメータバリデーション
@app.delete("/{user_id}")
async def delete_user(user_id: int):

    # 削除後のディクショナリを取得（実際に削除はしない）
    res_users = list(filter(lambda user: user['user_id'] != user_id, users))
    return JSONResponse(status_code=200, content=res_users)

引数のパスパラメータに型ヒントをつけることでパスパラメータのバリデーションができる。
この場合、引数がintにしてあるため、文字列が設定されている場合、バリデーションエラーになる。

culr


$ curl -i -X DELETE http://localhost:8000/aaa
HTTP/1.1 422 Unprocessable Entity
date: Thu, 19 Sep 2024 12:26:07 GMT
server: uvicorn
content-length: 152
content-type: application/json

{"detail":[{"type":"int_parsing","loc":["path","user_id"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"aaa"}]}

クエリパラメータ

クエリパラメータに対しても同様にバリデーションする

main.py


from fastapi import FastAPI
from fastapi.responses import JSONResponse
import copy
from typing import Optional

app = FastAPI()

# ディクショナリ
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}
]

# ageでフィルタリングして取得
# クエリパラメータバリデーション
@app.get("/", status_code=200)
async def get_users(age: Optional[int] = None):
    res_users = copy.deepcopy(users)

    # ageが指定された場合のみフィルタリング
    if age is not None:
        filtered_users = [user for user in res_users if user['age'] == age]
        return filtered_users if filtered_users else JSONResponse(status_code=404, content={"message": "No users found with the specified age"})

    # ageが指定されていない場合は全ユーザーを返す
    return JSONResponse(status_code=200, content=res_users)

引数のクエリパラメータに型ヒントをつけることでクエリパラメータのバリデーションができる。
この場合、引数がintにしてあるため、文字列が設定されている場合、バリデーションエラーになる。

culr


$ curl -i -X GET http://localhost:8000/?age=bb
HTTP/1.1 422 Unprocessable Entity
date: Thu, 19 Sep 2024 12:29:38 GMT
server: uvicorn
content-length: 148
content-type: application/json

{"detail":[{"type":"int_parsing","loc":["query","age"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"bb"}]}

デフォルトのエラーメッセージが出力される

リクエストボディ

リクエストボディのバリデーションはpydanticのデータモデルクラスを使って行う。

main.py


from fastapi import FastAPI
from fastapi.responses import JSONResponse
import copy
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# ディクショナリ
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}
]

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# 登録
# リクエストボディバリデーション
@app.post("/")
async def post_user(user: User):
    res_users = copy.deepcopy(users)

    # Userインスタンスをdictに変換して追加
    res_users.append(user.model_dump())
    return JSONResponse(status_code=201, content=res_users)

引数の型ヒントにpydanticのデータモデルクラス(User)を指定することで
リクエストボディをUserクラスでバリデーションすることができる

バリデーションが成功するとpost_user関数の引数userはUserクラスのインスタンスに変換されて渡されるので
ディクショナリに追加する時にmodel_dump()で辞書型に変換してやる必要がある。
※変換しないとJSON変換エラーになる

正常系（バリデーション成功）

curl


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": 30}'
HTTP/1.1 201 Created
date: Thu, 19 Sep 2024 12:53:19 GMT
server: uvicorn
content-length: 200
content-type: application/json

[{"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},{"user_id":5,"name":"nakamura","age":30}]

バリデーションが成功した場合は想定通りの値が返ってくる

異常系（バリデーション失敗）

curl


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": aa}'
HTTP/1.1 422 Unprocessable Entity
date: Thu, 19 Sep 2024 12:56:21 GMT
server: uvicorn
content-length: 125
content-type: application/json

{"detail":[{"type":"json_invalid","loc":["body",42],"msg":"JSON decode error","input":{},"ctx":{"error":"Expecting value"}}]}

ageに文字列を設定しているのでリクエストボディのバリデーションエラーになっている

レスポンスバリデーション

レスポンスのバリデーションもpydanticのデータモデルクラスを使って行う。

レスポンスバリデーションエラー（レスポンスデータが指定されたモデルに一致しない場合）は、
デフォルトでは「500 Internal Server Error」を返す。
レスポンスバリデーションエラーはアプリケーション内部のエラーとして扱われる。

またレスポンスのバリデーションはバリデーションを

自動バリデーション
手動バリデーション

のどちらでも行うことができる。

またバリデーション対象が

データモデルクラスのインスタンス
辞書型
JSON

のどれでもバリデーションできる※実装方法は少し変わる

ただ注意点としては

JSONResponseクラス
Responseクラス

を使ってレスポンスを作る場合は自動バリデーションは無効になるので
手動バリデーションを使う必要がある。
※pydanticに基づく構造を提供せず直接レスポンスデータを返すため無効になる

パターン別に実装して確認する

自動バリデーション

データモデルクラスを使ってレスポンスデータを
自動バリデーションする。

自動バリデーションはエンドポイントからレスポンスが返されるときに自動で
実行されるため、response_modelを引数として設定するだけで特に明示的に処理を書かなくていい。
※FastApiがレスポンス返却時、デフォルトで辞書型をJSONにシリアライズするのでそのタイミング実行される

FastAPI はエンドポイントが返すオブジェクトを
response_modelに指定されたデータモデルクラスに従ってシリアライズし、バリデーションする

バリデーション対象がデータモデルクラスのインスタンス

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスのインスタンス
user = User(user_id=1, name="Tujimura", age=11)

@app.get("/", response_model=User)
async def get_user():

    # 文字列に変更
    user.age="aaa"

    # 辞書型して自動バリデーション実行
    return user.model_dump()

バリデーションエラーにするためにageを文字列に変更
自動バリデーションは対象データを辞書型にする必要あり
レスポンスデータを辞書型に変換
自動バリデーションはreturnで実行される

実行すると500エラーになる

実行


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Fri, 20 Sep 2024 16:53:38 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

logを見るとResponseValidationErrorが発生する

log


fastapi.exceptions.ResponseValidationError: 1 validation errors:
  {'type': 'int_parsing', 'loc': ('response', 'age'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'aaa'}

※Uvicornで実行しているので標準出力に表示される

注意点としてはレスポンス返却時に自動バリデーションを行いResponseValidationErrorを
発生させるには返却時にmodel_dump()でインスタンスを辞書型に変換する必要がある
しない場合は200OKで返却されてしまう

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスのインスタンス
user = User(user_id=1, name="Tujimura", age=11)

@app.get("/", response_model=User)
async def get_user():

    # 文字列に変更
    user.age="aaa"

    # インスタンスのままで辞書型に変換しない
    return user

これで実行すると

実行


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 200 OK
date: Fri, 20 Sep 2024 17:01:05 GMT
server: uvicorn
content-length: 43
content-type: application/json

{"user_id":1,"name":"Tujimura","age":"aaa"}

200OKでageが文字列でそのまま返却される

ログを見ると

ログ


/root/.cache/pypoetry/virtualenvs/fastapi-restapi-xS3fZVNL-py3.12/lib/python3.12/site-packages/pydantic/type_adapter.py:451: UserWarning: Pydantic serializer warnings:
  Expected `int` but got `str` with value `'aaa'` - serialized value may not be as expected
  return self.serializer.to_python(
INFO:     172.18.0.1:45264 - "GET / HTTP/1.1" 200 OK

aaaが文字列である旨の警告はでている。

バリデーション対象がインスタンスの場合は注意する

バリデーション対象が辞書型

レスポンスバリデーションが実行されるタイミングは同様に
エンドポイントからレスポンスが返されるときになる。

レスポンスデータは既に辞書型であるため変換は不要。

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# ディクショナリ
user = {"user_id": 1, "name": "Tujimura", "age": 11}

@app.get("/", response_model=User)
async def get_user():

    # 文字列に変更
    user['age']="aaa"

    # 自動バリデーション
    return user

バリデーションエラーにするためにageを文字列に変更
自動バリデーションはreturnで実行される

実行すると500エラーになる

実行


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Fri, 20 Sep 2024 16:53:38 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

logを見るとResponseValidationErrorが発生する

log


fastapi.exceptions.ResponseValidationError: 1 validation errors:
  {'type': 'int_parsing', 'loc': ('response', 'age'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'aaa'}

※Uvicornで実行しているので標準出力に表示される

バリデーション対象がJSON

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
import json

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# JSON形式の文字列
user_json = '''
{
    "user_id": 1,
    "name": "Tujimura",
    "age": "aa"
}
'''

@app.get("/", response_model=User)
async def get_user():
    # 辞書型に変換して自動バリデーション実行    
    return json.loads(user_json)

バリデーションエラーにすためageを文字列にする
自動バリデーションは対象データが辞書型にする必要あり
レスポンスデータを辞書型に変換
自動バリデーションはreturnで実行される

実行すると500エラーになる

実行


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Fri, 20 Sep 2024 16:53:38 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

logを見るとResponseValidationErrorが発生する

log


fastapi.exceptions.ResponseValidationError: 1 validation errors:
  {'type': 'int_parsing', 'loc': ('response', 'age'), 'msg': 'Input should be a valid integer, unable to parse string as an integer', 'input': 'aa'}

手動バリデーション

手動バリデーションは自動バリデーションと違い明示的に
バリデーションを実行する。
手動で行うため自動で使うresponse_modelの設定は不要

具体的には

手動バリデーション処理抜粋


# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# モデルを使って手動でバリデーション
# バリデーション対象が辞書型の場合
validated_user = User(**user)

# バリデーション対象がJSONの場合
validated_user = User.model_validate_json(**user_json)

のようにして明示的にバリデーションを行う。
バリデーション対象が

辞書型
JSON

にどちらかより二つの方法がある

どちらもバリデーションが成功するとvalidated_userにはUserクラスのインスタンスが格納される
※辞書型やJSONではなくなるので注意！！！

バリデーション対象がデータモデルクラスのインスタンス

手動バリデーションでバリデーションエラーになる場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスのインスタンス
user = User(user_id=1, name="Tujimura", age=11)

@app.get("/")
async def get_user():

    # 文字列に変更
    user.age = "aaa"

    # モデルを使って手動でバリデーション (辞書型に変換してから)
    # ここでValidationErrorが発生する
    validated_user = User(**user.model_dump())

    return JSONResponse(status_code=200, content=validated_user.model_dump())

バリデーション実行時、引数は辞書型である必要があるため変換する
手動バリデーションでValidationErrorが発生し、レスポンスが返される
returnまで処理はこない

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Sat, 21 Sep 2024 12:26:43 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

ValidationErrorが起こっているので500エラーが返却される

ログを見ると

ログ


pydantic_core._pydantic_core.ValidationError: 1 validation error for User
age
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='aaa', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/int_parsing

バリデーションが失敗している

バリデーションが成功した場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスのインスタンス
user = User(user_id=1, name="Tujimura", age=11)

@app.get("/")
async def get_user():

    # 文字列に変更
    # user.age = "aaa"

    # モデルを使って手動でバリデーション (辞書型に変換してから)
    # バリデーションが成功した場合、インスタンスに変換される
    validated_user = User(**user.model_dump())

    # インスタンスから辞書型に変換
    return JSONResponse(status_code=200, content=validated_user.model_dump())

バリデーション実行時、引数は辞書型である必要があるため変換する
JSONResponseはインスタンスは自動シリアライズしてくれないので、辞書型に変換してやる

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 200 OK
date: Sat, 21 Sep 2024 12:28:39 GMT
server: uvicorn
content-length: 40
content-type: application/json

{"user_id":1,"name":"Tujimura","age":11}

正常に返却される

バリデーション対象が辞書型

バリデーション対象が辞書型の場合は変換が少なくて済む。
まずはバリデーションエラーの場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスの辞書
user = {"user_id": 1, "name": "Tujimura", "age": 11}

@app.get("/")
async def get_user():

    # 文字列に変更
    user["age"] = "aaa"

    # モデルを使って手動でバリデーション
    # ここでValidationErrorが発生する
    validated_user = User(**user)

    # インスタンスから辞書型に変換
    return JSONResponse(status_code=200, content=validated_user.model_dump())

手動バリデーションでValidationErrorが発生し、レスポンスが返される
returnまで処理はこない

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Sat, 21 Sep 2024 12:33:17 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

500エラーになる

ログを見ると

ログ


pydantic_core._pydantic_core.ValidationError: 1 validation error for User
age
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='aaa', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/int_parsing

バリデーションエラーになっているのがわかる

バリデーションが成功する場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスの辞書
user = {"user_id": 1, "name": "Tujimura", "age": 11}

@app.get("/")
async def get_user():

    # 文字列に変更
    # user["age"] = "aaa"

    # モデルを使って手動でバリデーション
    # バリデーション成功時、インスタンスに変換される
    validated_user = User(**user)

    # インスタンスから辞書型に変換
    return JSONResponse(status_code=200, content=validated_user.model_dump())

手動バリデーション成功時にインスタンス変換される
JSONResponseはインスタンスは自動シリアライズしてくれないので、辞書型に変換してやる

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 200 OK
date: Sat, 21 Sep 2024 12:28:39 GMT
server: uvicorn
content-length: 40
content-type: application/json

{"user_id":1,"name":"Tujimura","age":11}

正常に返却される

バリデーション対象がJSONの場合

手動バリデーションでバリデーションエラーになる場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# JSON形式の文字列
user_json = '''
{
    "user_id": 1,
    "name": "Tujimura",
    "age": "aa"
}
'''

@app.get("/")
async def get_user():
    # model_validate_jsonを使ってJSON文字列からバリデーションと変換
    user = User.model_validate_json(user_json)

    # 辞書型に変換して返却
    return JSONResponse(status_code=200, content=user.model_dump())

手動バリデーションでValidationErrorが発生し、レスポンスが返される
returnまで処理はこない

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 500 Internal Server Error
date: Sat, 21 Sep 2024 12:26:43 GMT
server: uvicorn
content-length: 21
content-type: text/plain; charset=utf-8

Internal Server Error

ValidationErrorが起こっているので500エラーが返却される

ログを見ると

ログ


pydantic_core._pydantic_core.ValidationError: 1 validation error for User
age
  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='aaa', input_type=str]
    For further information visit https://errors.pydantic.dev/2.9/v/int_parsing

バリデーションが失敗している

バリデーションが成功した場合

main.py


from fastapi import FastAPI
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# JSON形式の文字列
user_json = '''
{
    "user_id": 1,
    "name": "Tujimura",
    "age": 11
}
'''

@app.get("/")
async def get_user():
    # model_validate_jsonを使ってJSON文字列からバリデーションと変換
    user = User.model_validate_json(user_json)

    # 辞書型に変換して返却
    return JSONResponse(status_code=200, content=user.model_dump())

JSONResponseはインスタンスは自動シリアライズしてくれないので、辞書型に変換してやる

実行すると

curl


$ curl -i -X GET http://localhost:8000/
HTTP/1.1 200 OK
date: Sat, 21 Sep 2024 12:28:39 GMT
server: uvicorn
content-length: 40
content-type: application/json

{"user_id":1,"name":"Tujimura","age":11}

正常に返却される

Responesクラスを使った場合

Responesクラスは引数contentにはシリアライズしたものを渡さないと
いけないので下記のようになる。

main.py


from fastapi import FastAPI, Response
from typing import Optional
from pydantic import BaseModel
from fastapi.responses import JSONResponse

app = FastAPI()

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# Userクラスの辞書
user = {"user_id": 1, "name": "Tujimura", "age": 11}

@app.get("/")
async def get_user():

    # 文字列に変更
    # user["age"] = "aaa"

    # モデルを使って手動でバリデーション
    # バリデーション成功時、インスタンスに変換される
    validated_user = User(**user)

    # インスタンスからJSONに変換
    return Response(status_code=201, content=validated_user.model_dump_json(), media_type="application/json")

contentにはJSONに変換してから設定する

バリデーション要点まとめ

少し複雑なので要点をまとめる

自動バリデーションと手動バリデーションがある
自動バリデーション
- return時にレスポンスデータを自動でバリデーションする
- バリデーション対象データは辞書型である必要あり※違う場合は変換する
- JSONResponseとResponseでは自動バリデーションできない
手動バリデーション
- 手動バリデーションの方法は対象が辞書とJSONの場合で2パターンある
- 手動バリデーションが成功すると、データモデルクラスのインスタンスに変換され返却される
- JSONResponseとResponseと併用が可能。※手動バリデーションした後に使う

となる。

データシリアライゼーション・デシリアライゼーション

pydanticを使ってデータのシリアライズとデシリアライズを行うことができる。
上記で使ってるが一応まとめておく

シリアライズとは？

シリアライズはモデルのインスタンスを「辞書」や「JSON」などの形式に
変換すること

model_dump() ~モデルインスタンス → 辞書~

serialization.py


from pydantic import BaseModel

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

# Pydanticモデルのインスタンス
user = User(user_id=1, name="Charlie", age=40)

# Pydanticモデル自体の型も確認
print(f"ユーザモデルの型: {type(user)}")
# 出力: ユーザモデルの型: <class '__main__.User'>

# シリアライズ: Pydanticモデルを辞書に変換
user_dict = user.model_dump()
print(user_dict)
print(f"型: {type(user_dict)}")  # 型を出力
# 出力: {'user_id': 1, 'name': 'Charlie', 'age': 40}
# 型: <class 'dict'>

model_dump_json ~モデルインスタンス → JSON~

serialization.py


from pydantic import BaseModel

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

# Pydanticモデルのインスタンス
user = User(user_id=1, name="Charlie", age=40)

# Pydanticモデル自体の型も確認
print(f"ユーザモデルの型: {type(user)}")
# 出力: ユーザモデルの型: <class '__main__.User'>

# シリアライズ: PydanticモデルをJSON形式に変換
user_json = user.model_dump_json()
print(user_json)
print(f"型: {type(user_json)}")  # 型を出力
# 出力: {"user_id": 1, "name": "Charlie", "age": 40}
# 型: <class 'str'>

デシリアライズとは？

シリアライズとは逆で「辞書」や「JSON」などのデータ形式からPydanticのモデルにデータを読み込む

データモデルクラス(**対象辞書型データ) ~辞書型データ → モデルインスタンス~

deserialization.py


from pydantic import BaseModel

# Pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: int

# 辞書を使ったデシリアライズ（バリデーションを伴う）
user_data = {"user_id": 1, "name": "Charlie", "age": 40}
validated_user = User(**user_data)  # デシリアライズ
print(validated_user)
print(f"型: {type(validated_user)}")  # 型を出力
# 出力: user_id=1 name='Charlie' age=40
# 型: <class '__main__.User'>

バリデーションも同時に実施される

データモデルクラス.model_validate_json(JSONデータ) ~JSONデータ → モデルインスタンス~

deserialization.py


from pydantic import BaseModel

# Pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: int

# JSON文字列を使ったデシリアライズ（バリデーションを伴う）
user_json = '{"user_id": 2, "name": "Alice", "age": 30}'
validated_user_from_json = User.model_validate_json(user_json)  # デシリアライズ
print(validated_user_from_json)
print(f"型: {type(validated_user_from_json)}")  # 型を出力
# 出力: user_id=2 name='Alice' age=30
# 型: <class '__main__.User'>

バリデーションも同時に実施される
model_validate_jsonはクラスメソッド

バリデーションのエラーハンドリング

バリデーションのエラーハンドリング方法もまとめておく。

バリデーションで発生する例外は

RequestValidationError（リクエスト自動バリデーションで発生）
ResponseValidationError（レスポンス自動バリデーションで発生）
ValidationError（手動バリデーションで発生）

の3つになる

例外の補足方法

自動バリデーションについて

自動バリエーションの

RequestValidationError（リクエスト自動バリデーションで発生）
ResponseValidationError（レスポンス自動バリデーションで発生）

はFastApiの内部で自動的に処理されるため
「try~exept」では補足することができないのでエラーハンドラーを使って補足する

手動バリデーションについて

手動バリデーションは「try~exept」では補足が可能。
ただ発生する例外は「ValidationError」なのでrequestかresponseなのか判断できない
※requestは自動ですると思うので「RequestValidationError」か「ValidationError」かで判断する

エラーハンドラー実装

エラーハンドラーを実装してみる
実装パターンとして

リクエスト自動バリデーション + レスポンス自動バリデーション
リクエスト自動バリデーション + レスポンス手動バリデーション

のパターンを実装する

リクエスト自動バリデーション + レスポンス自動バリデーション

自動バリデーションなのでエラーハンドラーでバリデーションエラーを捕まえる

main.py


from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError, ResponseValidationError
from pydantic import BaseModel, ValidationError
import copy
from typing import List, Optional

app = FastAPI()

# ディクショナリ
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}
]

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# データモデルリストの定義
class UsersList(BaseModel):
    users: Optional[List[User]] = []

# リクエストバリデーションエラーハンドラー
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "message": 'RequestValidationError',
            "detail": exc.errors(),
            "body": exc.body
        }
    )

# レスポンスバリデーションエラーハンドラー
@app.exception_handler(ResponseValidationError)
async def response_validation_exception_handler(request: Request, exc: ValidationError):
    return JSONResponse(
        status_code=500,
        content={
            "message": 'ResponseValidationError',
            "detail": exc.errors()
        }
    )

# 登録
# リクエストボディ自動バリデーション
@app.post("/", status_code=201, response_model=List[User])
async def post_user(user: User):
    res_users = copy.deepcopy(users)

    # 文字列に変更
    # user.age = "aaa"

    # リクエストバリデーション実施後、Userインスタンスになっているため
    # Userインスタンスをdictに変換して追加
    res_users.append(user.model_dump())

    # 返却時にレスポンス自動バリデーション実行
    return res_users

リクエスト自動バリデーションpost_userで引数を受けた時に実行される
リクエストバリデーションが成功した場合はインスタンスになるので辞書型に変換してListに追加
return部分でレスポンス自動バリエーションが実行される
リクエスト自動バリデーションでエラーが発生 →　リクエストバリデーションエラーハンドラーで処理する
レスポンス自動バリデーションでエラーが発生 →　レスポンスバリデーションエラーハンドラーで処理する

実行~バリエーションエラーなし~

curlでエラーが発生しないように実行してみる

正常


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": 11}'
HTTP/1.1 201 Created
date: Sun, 22 Sep 2024 12:15:20 GMT
server: uvicorn
content-length: 200
content-type: application/json

[{"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},{"user_id":5,"name":"nakamura","age":11}]

正常にレスポンスが返ってくる

実行~リクエストバリデーションエラー~

ageを文字列にしてリクエストする

リクエストバリデーションエラー


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": aa}'
HTTP/1.1 422 Unprocessable Entity
date: Sun, 22 Sep 2024 12:15:39 GMT
server: uvicorn
content-length: 223
content-type: application/json

{"message":"RequestValidationError","detail":[{"type":"json_invalid","loc":["body",42],"msg":"JSON decode error","input":{},"ctx":{"error":"Expecting value"}}],"body":"{\"user_id\": 5, \"name\": \"nakamura\", \"age\": aa}"}

RequestValidationErrorが発生する

実行~レスポンスバリデーションエラー~

ソースのコメントアウトしている「user.age = "aaa"」を解除して
レスポンスのageを文字列にする

レスポンスバリデーションエラー


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": 11}'
HTTP/1.1 500 Internal Server Error
date: Sun, 22 Sep 2024 12:16:00 GMT
server: uvicorn
content-length: 190
content-type: application/json

{"message":"ResponseValidationError","detail":[{"type":"int_parsing","loc":["response",4,"age"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"aaa"}]}

ResponseValidationErrorが発生する

リクエスト自動バリデーション + レスポンス手動バリデーション

レスポンスは手動でバリデーションする
リクエスト自動バリデーションは同じなので解説を割愛する

main.py


from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError, ResponseValidationError
from pydantic import BaseModel, ValidationError
import copy
from typing import List, Optional

app = FastAPI()

# ディクショナリ
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}
]

# pydanticデータモデル
class User(BaseModel):
    user_id: int
    name: str
    age: Optional[int]

# データモデルリストの定義
class UsersList(BaseModel):
    users: Optional[List[User]] = []

# リクエストバリデーションエラーハンドラー
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={
            "message": 'RequestValidationError',
            "detail": exc.errors(),
            "body": exc.body
        }
    )

# 登録
# リクエストボディ自動バリデーション
@app.post("/")
async def post_user(user: User):
    res_users = copy.deepcopy(users)

    # リクエストバリデーション実施後、Userインスタンスになっているため
    # Userインスタンスをdictに変換して追加
    res_users.append(user.model_dump())

    # データモデルクラスリストインスタンス作成
    validate_target_users = UsersList(users=res_users)

    # 文字列に変更
    # validate_target_users.users[0].age = "aaa"

    # 手動バリデーション実行をtry-exceptで囲む
    try:
        validated_users = UsersList(**validate_target_users.model_dump())
    except ValidationError as exc:
        # ValidationError発生時にエラーレスポンスを返す
        return JSONResponse(
            status_code=500,
            content={
                "message": 'ValidationError',
                "detail": exc.errors()
            }
        )

    # 正常時のレスポンス返却
    return JSONResponse(status_code=201, content=validated_users.model_dump())

レスポンス返却前に手動でバリデーションする
手動の場合はValidationErrorが発生するのでtry~exeptで捕まえる
try~exeptを使いたくない場合はValidationErrorのエラーハンドラー関数で捕まえることもできる

実行~バリエーションエラーなし~

curlでエラーが発生しないように実行してみる

正常


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": 11}'
HTTP/1.1 201 Created
date: Sun, 22 Sep 2024 12:47:12 GMT
server: uvicorn
content-length: 210
content-type: application/json

{"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},{"user_id":5,"name":"nakamura","age":11}]}

正常にレスポンスが返ってくる

実行~レスポンスバリデーションエラー~

ソースのコメントアウトしている「validate_target_users.users[0].age = "aaa"」を解除して
レスポンスのageを文字列にする

レスポンスバリデーションエラー


$ curl -i -X POST "http://127.0.0.1:8000/" -H "Content-Type: application/json" -d '{"user_id": 5, "name": "nakamura", "age": 11}'
HTTP/1.1 500 Internal Server Error
date: Sun, 22 Sep 2024 12:48:07 GMT
server: uvicorn
content-length: 233
content-type: application/json

{"message":"ValidationError","detail":[{"type":"int_parsing","loc":["users",0,"age"],"msg":"Input should be a valid integer, unable to parse string as an integer","input":"aaa","url":"https://errors.pydantic.dev/2.9/v/int_parsing"}]}

ValidationErrorが発生する

まとめ

FastApiのリクエストとレスポンスのバリデーション方法と
そのエラーハンドリングについて実装をしながらまとめてみた。

バリデーション対象が

インスタンス
辞書
JSON

のどれかによって実装が変わるので、対象のデータ型は意識しておく必要がある。

また、FastApiが自動で行ってくれるシリアライズについても知らないと
動きがわからなくなるので注意。

当記事ではバリデーション部分のみをピックアップしているが
全体的なREST APIの実装や動かし方については下記記事でまとめているので
よければご参照ください

参照

response_modelとは【FastAPI】

【FastApi × Poetry】pydanticでリクエスト・レスポンスをバリデーションする

動かして学ぶ！Python FastAPI開発入門 （NEXT ONE） [ 中村 翔 ]

pydanticとは？

なぜFastApiでpydanticを使うのか？

pydanticでできること

リクエストバリデーション

パスパラメータ

クエリパラメータ

リクエストボディ

正常系（バリデーション成功）

異常系（バリデーション失敗）

レスポンスバリデーション

自動バリデーション

バリデーション対象がデータモデルクラスのインスタンス

バリデーション対象が辞書型

バリデーション対象がJSON

手動バリデーション

バリデーション対象がデータモデルクラスのインスタンス

バリデーション対象が辞書型

バリデーション対象がJSONの場合

Responesクラスを使った場合

バリデーション要点まとめ

データシリアライゼーション・デシリアライゼーション

シリアライズとは？

model_dump() ~モデルインスタンス → 辞書~

model_dump_json ~モデルインスタンス → JSON~

デシリアライズとは？

データモデルクラス(**対象辞書型データ) ~辞書型データ → モデルインスタンス~

データモデルクラス.model_validate_json(JSONデータ) ~JSONデータ → モデルインスタンス~

バリデーションのエラーハンドリング

例外の補足方法

自動バリデーションについて

手動バリデーションについて

エラーハンドラー実装

リクエスト自動バリデーション + レスポンス自動バリデーション

実行~バリエーションエラーなし~

実行~リクエストバリデーションエラー~

実行~レスポンスバリデーションエラー~

リクエスト自動バリデーション + レスポンス手動バリデーション

実行~バリエーションエラーなし~

実行~レスポンスバリデーションエラー~

まとめ

参照

新着記事

動かして学ぶ！Python FastAPI開発入門（NEXT ONE） [ 中村翔 ]