当サイトは、アフィリエイト広告を利用しています
PythonのFastApiで作ったREST APIでリクエストとレスポンスを
バリデーションライブラリであるpydanticを使って行ってみる。
タイトルではリクエスト・レスポンスのバリデーションと書いたが
その他にpydanticを使ってできることもまとめておく。
※バリデーションエラーのエラーハンドリングも
FastApiのレスポンス生成については下記でまとめている
pydanticは型ヒントを使用してデータのバリデーションを行うライブラリ。
詳しくは下記参照
FastApiとPydanticは親和性が非常に高いため。
というかFastApiがPydanticをネイティブに統合している。
FastApi自体がPydanticに依存しており、FastApiをインストールすると、
Pydanticも自動的にインストールされる。
つまり、FastApiを使う場合はPydanticはインストールなしの使うことができるので バリデーションライブラリとして他のものを使う必要はない
またPydanticは型チェックやバリデーションの処理が非常に効率的で、
高速なパフォーマンスなのでリアルタイム性が求められるAPI開発でもパフォーマンスが損なわれない。
ちなみにFlaskではデフォルトでのバリデーション機能はないのでpydantic等の
ライブラリを使う必要があり、使ったとしてはFastApiほど簡単には実装できない。
※ネイティブに統合してるだけあり、FastApiの方が圧倒的に使いやすい
flaskでpydanticを使ってバリデーションする方法は一応下記記事で紹介している
pydanticを使うことで
ができる。
実際にFastApiでREST APIを作って動作を確認する
リクエストバリデーション使用される関数
については下記で細かくまとめている
リクエストバリデーションをpydanticを使ってやってみる。
pydanticによってリクエストバリデーションが実行されるタイミングは
になる。
@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」を返す
リクエストは分けると
になるので、それぞれについてバリデーションしてみる
パスパラメータに対するバリデーションを行う
from fastapi import FastAPIfrom fastapi.responses import JSONResponseapp = 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 = Path(..., description="ユーザーID", ge=1)):# 削除後のディクショナリを取得(実際に削除はしない)res_users = list(filter(lambda user: user['user_id'] != user_id, users))return JSONResponse(status_code=200, content=res_users)
引数のパスパラメータに型ヒントをつけることでパスパラメータのバリデーションができる。
引数部分に関して詳しく解説する
型ヒント。
user_id は整数 (int) 型であることを示している。
FastAPI は型ヒントを利用してリクエストのバリデーションや自動ドキュメント生成を行う。
PathはFastAPIが用意しているパラメータ関数。
Pathを使うことで「これは パスパラメータ です」と宣言できる。
「...」は必須を意味する特殊値(省略不可ということ)
OpenAPIドキュメント(Swagger UI など)に表示される説明文。
API の利用者に「この値は何か?」を説明するために使う。
バリデーション制約。 ge は「greater than or equal(以上)」の意味。
user_id >= 1 でなければリクエストは自動的に 422 エラーになる。
この場合、引数がintにしてあるため、文字列が設定されている場合、バリデーションエラーになる。
$ curl -i -X DELETE http://localhost:8000/aaaHTTP/1.1 422 Unprocessable Entitydate: Thu, 19 Sep 2024 12:26:07 GMTserver: uvicorncontent-length: 152content-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"}]}
下記のように省略して書くこともできる。
@app.get("/users/{user_id}")def get_user(user_id: int): # 型だけ指定# 削除後のディクショナリを取得(実際に削除はしない)res_users = list(filter(lambda user: user['user_id'] != user_id, users))return JSONResponse(status_code=200, content=res_users)
ただし、説明や制約は追加できない
クエリパラメータに対しても同様にバリデーションする
from fastapi import FastAPI, Queryfrom fastapi.responses import JSONResponseimport copyapp = 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: int | None = Query(None, description="ユーザーの年齢でフィルタ")):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_usersif filtered_userselse JSONResponse(status_code=404,content={"message": "No users found with the specified age"},))# ageが指定されていない場合は全ユーザーを返すreturn JSONResponse(status_code=200, content=res_users)
引数のクエリパラメータに型ヒントをつけることでクエリパラメータのバリデーションができる。
引数部分に関して詳しく解説する
まとめると
ということ。
※Query(...) にすれば必須パラメータにできる。その場合は「| None」は消さないと矛盾する
pythonでは引数にデフォルト値をつけない場合、その引数は必須引数になる。
またデフォルト値に、Noneを指定する場合は、オプショナル型にする必要がある。
オプショナルの書き方は下記の通り(3パターンある)
# 新しい書き方(Python 3.10+ 推奨)age: int | None = Query(None)# 従来の書き方(古いバージョンでもOK)from typing import Optionalage: Optional[int] = Query(None)# さらに明示的に Union を使う書き方from typing import Unionage: Union[int, None] = Query(None)
この場合、引数がintにしてあるため、文字列が設定されている場合、バリデーションエラーになる。
$ curl -i -X GET http://localhost:8000/?age=bbHTTP/1.1 422 Unprocessable Entitydate: Thu, 19 Sep 2024 12:29:38 GMTserver: uvicorncontent-length: 148content-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の データモデルクラスを使って行う。
from fastapi import FastAPIfrom fastapi.responses import JSONResponseimport copyfrom typing import Optionalfrom pydantic import BaseModelapp = 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: intname: strage: Optional[int]# 登録# リクエストボディバリデーション@app.post("/")async def post_user(user: User = Body(..., embed=False, description="新しいユーザーを登録します")):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変換エラーになる
Body()の引数embed関して詳しく解説する
※他はパスパラメータと同様なので割愛
embed=True / False はFastAPIがリクエストJSONをPydanticモデルにマッピングする方法を制御している
例えばフロントがリクエストボディ部分を
{ "user_id":1, "name":"Taro", "age":30 }
のようにUserのフィールドを直接送る形式だった場合はembed=Falseにする
またフロントがリクエストボディ部分を
{ "user": { "user_id":1, "name":"Taro", "age":30 } }
のようにUserを単一キーでラップして送る形式だった場合はembed=Trueにする
つまり、FastAPI側の処理としては
のようにしてフロントの送信形式によっていPydanticモデルにマッピングする方法を制御できる
$ 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 Createddate: Thu, 19 Sep 2024 12:53:19 GMTserver: uvicorncontent-length: 200content-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 -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 Entitydate: Thu, 19 Sep 2024 12:56:21 GMTserver: uvicorncontent-length: 125content-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」 を返す。
レスポンスバリデーションエラーはアプリケーション内部のエラーとして扱われる。
またレスポンスのバリデーションはバリデーションを
のどちらでも行うことができる。
またバリデーション対象が
のどれでもバリデーションできる※実装方法は少し変わる
ただ注意点としては
を使ってレスポンスを作る場合は自動バリデーションは無効になるので
手動バリデーションを使う必要がある。
※pydanticに基づく構造を提供せず直接レスポンスデータを返すため無効になる
パターン別に実装して確認する
データモデルクラスを使ってレスポンスデータを
自動バリデーションする。
自動バリデーションはエンドポイントからレスポンスが返されるときに自動で
実行されるため、response_modelを引数として設定するだけで特に明示的に処理を書かなくていい。
※FastApiがレスポンス返却時、デフォルトで辞書型をJSONにシリアライズするのでそのタイミング実行される
FastAPI はエンドポイントが返すオブジェクトを
response_modelに指定されたデータモデルクラスに従ってシリアライズし、バリデーションする
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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()
実行すると500エラーになる
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Fri, 20 Sep 2024 16:53:38 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
logを見るとResponseValidationErrorが発生する
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で返却されてしまう
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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 OKdate: Fri, 20 Sep 2024 17:01:05 GMTserver: uvicorncontent-length: 43content-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 expectedreturn self.serializer.to_python(INFO: 172.18.0.1:45264 - "GET / HTTP/1.1" 200 OK
aaaが文字列である旨の警告はでている。
バリデーション対象がインスタンスの場合は注意する
レスポンスバリデーションが実行されるタイミングは同様に
エンドポイントからレスポンスが返されるときになる。
レスポンスデータは既に辞書型であるため変換は不要。
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: Optional[int]# ディクショナリuser = {"user_id": 1, "name": "Tujimura", "age": 11}@app.get("/", response_model=User)async def get_user():# 文字列に変更user['age']="aaa"# 自動バリデーションreturn user
実行すると500エラーになる
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Fri, 20 Sep 2024 16:53:38 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
logを見るとResponseValidationErrorが発生する
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で実行しているので標準出力に表示される
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelimport jsonapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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)
実行すると500エラーになる
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Fri, 20 Sep 2024 16:53:38 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
logを見るとResponseValidationErrorが発生する
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: intname: strage: Optional[int]# モデルを使って手動でバリデーション# バリデーション対象が辞書型の場合validated_user = User(**user)# バリデーション対象がJSONの場合validated_user = User.model_validate_json(**user_json)
のようにして明示的にバリデーションを行う。
バリデーション対象が
にどちらかより二つの方法がある
どちらもバリデーションが成功するとvalidated_userにはUserクラスのインスタンスが格納される
※辞書型やJSONではなくなるので注意!!!
手動バリデーションでバリデーションエラーになる場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Sat, 21 Sep 2024 12:26:43 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
ValidationErrorが起こっているので500エラーが返却される
ログを見ると
pydantic_core._pydantic_core.ValidationError: 1 validation error for UserageInput 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
バリデーションが失敗している
バリデーションが成功した場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 200 OKdate: Sat, 21 Sep 2024 12:28:39 GMTserver: uvicorncontent-length: 40content-type: application/json{"user_id":1,"name":"Tujimura","age":11}
正常に返却される
バリデーション対象が辞書型の場合は変換が少なくて済む。
まずはバリデーションエラーの場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Sat, 21 Sep 2024 12:33:17 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
500エラーになる
ログを見ると
pydantic_core._pydantic_core.ValidationError: 1 validation error for UserageInput 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
バリデーションエラーになっているのがわかる
バリデーションが成功する場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 200 OKdate: Sat, 21 Sep 2024 12:28:39 GMTserver: uvicorncontent-length: 40content-type: application/json{"user_id":1,"name":"Tujimura","age":11}
正常に返却される
手動バリデーションでバリデーションエラーになる場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 500 Internal Server Errordate: Sat, 21 Sep 2024 12:26:43 GMTserver: uvicorncontent-length: 21content-type: text/plain; charset=utf-8Internal Server Error
ValidationErrorが起こっているので500エラーが返却される
ログを見ると
pydantic_core._pydantic_core.ValidationError: 1 validation error for UserageInput 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
バリデーションが失敗している
バリデーションが成功した場合
from fastapi import FastAPIfrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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())
実行すると
$ curl -i -X GET http://localhost:8000/HTTP/1.1 200 OKdate: Sat, 21 Sep 2024 12:28:39 GMTserver: uvicorncontent-length: 40content-type: application/json{"user_id":1,"name":"Tujimura","age":11}
正常に返却される
Responesクラスは引数contentにはシリアライズしたものを渡さないと
いけないので下記のようになる。
from fastapi import FastAPI, Responsefrom typing import Optionalfrom pydantic import BaseModelfrom fastapi.responses import JSONResponseapp = FastAPI()# pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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")
少し複雑なので要点をまとめる
自動バリデーションと手動バリデーションがある
自動バリデーション
手動バリデーション
となる。
pydanticを使ってデータのシリアライズとデシリアライズを行うことができる。
上記で使ってるが一応まとめておく
シリアライズはモデルのインスタンスを「辞書」や「JSON」などの形式に
変換すること
from pydantic import BaseModelclass User(BaseModel):user_id: intname: strage: 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'>
from pydantic import BaseModelclass User(BaseModel):user_id: intname: strage: 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のモデルにデータを読み込む
from pydantic import BaseModel# Pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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'>
from pydantic import BaseModel# Pydanticデータモデルclass User(BaseModel):user_id: intname: strage: 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'>
バリデーションのエラーハンドリング方法もまとめておく。
バリデーションで発生する例外は
の3つになる
自動バリエーションの
はFastApiの内部で自動的に処理されるため
「try~exept」では補足することができないのでエラーハンドラーを使って補足する
手動バリデーションは「try~exept」では補足が可能。
ただ発生する例外は「ValidationError」なのでrequestかresponseなのか判断できない
※requestは自動ですると思うので「RequestValidationError」か「ValidationError」かで判断する
エラーハンドラーを実装してみる
実装パターンとして
のパターンを実装する
自動バリデーションなのでエラーハンドラーでバリデーションエラーを捕まえる
from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.exceptions import RequestValidationError, ResponseValidationErrorfrom pydantic import BaseModel, ValidationErrorimport copyfrom typing import List, Optionalapp = 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: intname: strage: 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 Createddate: Sun, 22 Sep 2024 12:15:20 GMTserver: uvicorncontent-length: 200content-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 Entitydate: Sun, 22 Sep 2024 12:15:39 GMTserver: uvicorncontent-length: 223content-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 Errordate: Sun, 22 Sep 2024 12:16:00 GMTserver: uvicorncontent-length: 190content-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が発生する
レスポンスは手動でバリデーションする
リクエスト自動バリデーションは同じなので解説を割愛する
from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom fastapi.exceptions import RequestValidationError, ResponseValidationErrorfrom pydantic import BaseModel, ValidationErrorimport copyfrom typing import List, Optionalapp = 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: intname: strage: 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())
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 Createddate: Sun, 22 Sep 2024 12:47:12 GMTserver: uvicorncontent-length: 210content-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 Errordate: Sun, 22 Sep 2024 12:48:07 GMTserver: uvicorncontent-length: 233content-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のリクエストとレスポンスのバリデーション方法と
そのエラーハンドリングについて実装をしながらまとめてみた。
バリデーション対象が
のどれかによって実装が変わるので、対象のデータ型は意識しておく必要がある。
また、FastApiが自動で行ってくれるシリアライズについても知らないと
動きがわからなくなるので注意。
当記事ではバリデーション部分のみをピックアップしているが
全体的なREST APIの実装や動かし方については下記記事でまとめているので
よければご参照ください