こんにちはスナックミー CTO の三好 (@miyoshihayato) です
FastAPIを開発するときに独自でエラーハンドリングを設定する方法を書いていきたいと思います。

FastAPIの採用背景は以下をご覧ください

リプレイスの場合、エラーハンドリングはリプレイス前の仕様と同じにする必要があり、少なからずレスポンスはコントロールできる状態が必要でした。リプレイスではなく初めからFastAPIを採用したという方でも、日本語に対応できるのだろうか？エラーコードってどういう仕様なのだろうか？などカスタムしたくなるシーンは出てくると思います。

FastAPIのデフォルトのエラーハンドリング
独自定義したExceptionを利用
まとめ

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):
    """カスタム例外"""

    # デフォルトを401エラーとする
    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)