こんにちは スナックミー CTO の三好 (@miyoshihayato) です
FastAPIを開発するときに独自でエラーハンドリングを設定する方法を書いていきたいと思います。
FastAPIの採用背景は以下をご覧ください
labs.snaq.me
リプレイスの場合、エラーハンドリングはリプレイス前の仕様と同じにする必要があり、少なからずレスポンスはコントロールできる状態が必要でした。リプレイスではなく初めからFastAPIを採用したという方でも、日本語に対応できるのだろうか？エラーコードってどういう仕様なのだろうか？などカスタムしたくなるシーンは出てくると思います。
FastAPIのデフォルトのエラーハンドリング
パスがない場合
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
Request URL
http://127.0.0.1/hello_world
Response body
{
"detail": "Not Found"
}
パラメータ不足時のエラー
from fastapi import FastAPI, Form
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str = Form(..., title="名前", description="名前")
price: float = Form(..., title="金額", description="金額")
is_snaqme: bool = None
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.post("/items/{item_id}")
def update_item(item_id: int, item: Item):
return {"item_name": item.name, "item_id": item_id}
Request URL
http://127.0.0.1/items/1
Request body
{
"name": "パイナップル",
"is_snaqme": true
}
price の値が必要だがrequest bodyに入れない場合
Response body
{
"detail": [{
"loc": [
"body",
"price"
],
"msg": "field required",
"type": "value_error.missing"
}]
}
バリデーションエラー
from typing import Union
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
def read_items(q: Union[str, None] = Query(default=None, max_length=5)):
results = {"items": [{"item_id": "snaqme"}, {"item_id": "otumame"}]}
if q:
results.update({"q": q})
return results
q は最大文字数5文字のところ6文字で入れた場合
Request URL
http://127.0.0.1/items?q=123456
Response body
{
"detail": [{
"loc": [
"query",
"q"
],
"msg": "ensure this value has at most 5 characters",
"type": "value_error.any_str.max_length",
"ctx": {
"limit_value": 5
}
}]
}
このような Response body になっています。detail 直下に loc (エラー箇所), msg (メッセージ), type (エラータイプ) で構成されています。バリデーション設定によりその他のエラーメッセージも表示されています。
一方で以下のような
{
"error": {
"code": 401,
"message": "ログイン失敗しました"
}
}
のようにエラーコードとメッセージだけ出したい場合どのように行うとよいのでしょうか？
独自定義したExceptionを利用
import traceback
from fastapi import Request, Response, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
class CustomException(Exception):
"""カスタム例外"""
default_status_code = status.HTTP_401_UNAUTHORIZED
def __init__(
self,
msg: str,
status_code: int = default_status_code,
) -> None:
self.status_code = status_code
self.detail = {"error": {"code": status_code, "message": msg}}
class SystemException(Exception):
"""システム例外"""
def __init__(self, e: Exception) -> None:
self.exc = e
self.stack_trace = traceback.format_exc()
self.status_code = status.HTTP_500_INTERNAL_SERVER_ERROR
self.detail = {
"error": {
"code": status.HTTP_500_INTERNAL_SERVER_ERROR,
"message": "システムエラーが発生しました。",
}
}
class HttpRequestMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next) -> Response:
try:
response: Response = await call_next(request)
except CustomException as ce:
response = JSONResponse(ce.detail, status_code=ce.status_code)
except Exception as e:
se = SystemException(e)
response = JSONResponse(se.detail, status_code=se.status_code)
return response
CustomException のようにさまざまなパターンで例外処理を作成することが可能です。
呼び出し方は以下のようになります
from config.middleware import CustomException, HttpRequestMiddleware
from fastapi import FastAPI, Query
app = FastAPI()
@app.get("/items/")
async def read_items(q: Union[str, None] = Query(default=None, max_length=5)):
if q != "snaq.me":
rails CustomException(msg=f"not {q} but snaq.me")
return {"Hello": "World"}
app.add_middleware(HttpRequestMiddleware)
まとめ
このようにエラーハンドリングを自由にカスタムできチームとして思い通りのエラーハンドリングの参考になれば幸いです。
またエラーハンドリングはレスポンスに特化しましたが、バリデーションもFastAPI既存の仕様に従うかどうかでやり方が変わってきます。
次回以降もFastAPIで培ってきたこと発信します。
弊社では
「おやつと、世界を面白く。」
一緒に面白くしたい仲間をお待ちしてます！！
meety.net
engineers.snaq.me
