FastAPIでURLのバリデーションをpydanticを使って行う方法

• 概要
• 環境
• サンプルコード
• ポイント
• 実行

概要

カスタムリクエスト＋カスタムルーティングと組み合わせて実現します

環境

• macOS 11.6.8
• Python 3.10.2
• FastAPI 0.83.0

サンプルコード

"""カスタムルートをテストするモジュール."""
import ast
import json

from pydantic import BaseModel, BaseSettings, validator

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute

from typing import Callable


class Settings(BaseSettings):
    """一時的なデータ保存用のクラス."""

    message: str = ""
    hostname: str = ""


settings = Settings()


class RequestWithHostname(Request):
    """ホスト名をボディに必ず含むリクエスト."""

    async def body(self) -> bytes:
        """bodyメソッドのオーバライド."""
        if not hasattr(self, "_body"):
            body = await super().body()
            text_body = body.decode('utf-8')
            dict_body = ast.literal_eval(text_body)
            # url をセット
            url = self.url
            dict_body['hostname'] = url.hostname
            # bytes に再変換
            body = json.dumps(dict_body).encode('utf-8')
            self._body = body
        return self._body


class ContextIncludedRoute(APIRoute):
    """カスタムルートの定義."""

    def get_route_handler(self) -> Callable:
        """get_route_handlerのオーバライド."""
        # ルートハンドラの取得
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            """カスタムルートハンドラの定義.

            ここでルーティングごとに共通のロギングやヘッダのカスタマイズを行う.
            
            """
            request = RequestWithHostname(request.scope, request.receive)
            return await original_route_handler(request)

        # get_route_handlerは定義したカスタムルートハンドラを返却する
        return custom_route_handler


# アプリの作成とカスタムルートクラスの設定
app = FastAPI()
router = APIRouter(route_class=ContextIncludedRoute)


class Item(BaseModel):
    """pydanticを使ったリクエストモデル."""

    message: str
    hostname: str

    @validator('hostname')
    def is_fuga(cls, v, values, **kwargs):
        """ホスト名をチェック."""
        if v != 'fuga':
            raise ValueError('Hostname must be fuga')
        return v


# 定義したrouterを元に各種ルーティングを定義
@router.get("/")
async def get_msg():
    """メッセージを取得."""
    return {"message": settings.message, "hostname": settings.hostname}


@router.post("/")
def set_msg(item: Item):
    """メッセージの登録."""
    settings.message = item.message
    settings.hostname = item.hostname
    return {"message": item.message, "hostname": item.hostname}


# 定義したルーティングをアプリに登録する
app.include_router(router)

ポイント

• カスタムクラス＋カスタムルーティングを使ってボティの要素を追加する
• URLやホスト名の情報は Request オブジェクトから取得する
• カスタムクラスを扱う場合はボディ情報が bytes 型なので型変換が必要になる

実行

• pipenv run uvicorn app:app --reload

成功

• curl -v -XPOST localhost:8000 -H 'Host: fuga' -H 'content-type: application/json' -d '{"message":"hoge"}'

失敗

• curl -v -XPOST localhost:8000 -H 'Host: fuga2' -H 'content-type: application/json' -d '{"message":"hoge"}'