FastAPIのエラーハンドリング #Python

概要

FastAPIのエラーハンドリングについて
エントリーポイントと同じファイルに定義するかどうかで処理が変化
- 同じ場合はハンドラーを使用
- 違う場合はミドルウェアを使用

エントリーポイントと同じファイルで定義

ディレクトリ構造

fastapi_example
├── __init__.py
└── main.py         # FastAPIのエントリーポイント

実装例

fastapi_example/main.py

#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""FastAPIのエントリーポイント
"""

from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()


@app.exception_handler(Exception)
async def exception_handler(request, exc):
    return JSONResponse(
        status_code=status.HTTP_401_UNAUTHORIZED,
        content={"message": f"{type(exc)}"},
    )


@app.get("/exception")
async def exception():
    raise Exception

参考

Handling Errors

エントリーポイントと別ファイルで定義

ディレクトリ構造

fastapi_example
├── __init__.py
├── error.py        # エラー関連のモジュール
├── main.py         # FastAPIのエントリーポイント
└── middleware.py   # ミドルウェア関連のモジュール

実装例

fastapi_example/main.py

#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""FastAPIのエントリーポイント
"""

from fastapi import FastAPI

from fastapi_example.error import MyError
from fastapi_example.middleware import HttpRequestMiddleware

app = FastAPI()
app.add_middleware(HttpRequestMiddleware)


@app.get("/my")
async def my_error():
    try:
        int("hoge")
    except Exception as exc:
        raise MyError from exc

@app.get("/value")
async def value_error():
    int("hoge")

@app.get("/exception")
async def exception():
    raise Exception

fastapi_example/error

#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""エラー関連のモジュール
"""


class MyError(Exception):
    pass

fastapi_example/middleware

#!/usr/bin/env python3
# -*- coding: utf_8 -*-
"""ミドルウェア関連のモジュール
"""

from fastapi import Request, Response, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

from fastapi_example.error import MyError
import traceback

class HttpRequestMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next) -> Response:
        try:
            response: Response = await call_next(request)
        except MyError as exc:
            print(traceback.format_exc())
            response = JSONResponse(
                status_code=status.HTTP_401_UNAUTHORIZED,
                content={"message": f"{type(exc)}"},
            )
        except ValueError as exc:
            print(traceback.format_exc())
            response = JSONResponse(
                status_code=status.HTTP_402_PAYMENT_REQUIRED,
                content={"message": f"{type(exc)}"},
            )
        except Exception as exc:
            print(traceback.format_exc())
            response = JSONResponse(
                status_code=status.HTTP_400_BAD_REQUEST,
                content={"message": f"{type(exc)}"},
            )
        return response