概要
最初、各ルーティングの共通処理なのでミドルウェアがいいかなと思っていたのですがリクエストボディが扱えないようです
今回はカスタムルートを使ってリクエストとレスポンス情報をロギングする方法を紹介します
環境
- macOS 11.6.8
- Python 3.10.2
- FastAPI 0.83.0
サンプルコード
-
vim app.py
"""カスタムルートをテストするモジュール."""
from pydantic import BaseModel
from pydantic import BaseSettings
from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute
from typing import Callable
from uuid import uuid4
class Settings(BaseSettings):
"""一時的なデータ保存用のクラス."""
message: str = ""
settings = Settings()
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_id = str(uuid4())
# レスポンスオブジェクトの取得
response: Response = await original_route_handler(request)
# リクエストボディのロギング
if await request.body():
print(await request.body())
# カスタムヘッダ Request-ID の登録
response.headers["Request-ID"] = request_id
return response
# get_route_handlerは定義したカスタムルートハンドラを返却する
return custom_route_handler
# アプリの作成とカスタムルートクラスの設定
app = FastAPI()
router = APIRouter(route_class=ContextIncludedRoute)
class Item(BaseModel):
"""pydanticを使ったリクエストモデル."""
message: str
# 定義したrouterを元に各種ルーティングを定義
@router.get("/")
async def get_msg():
"""メッセージを取得."""
return {"message": settings.message}
@router.post("/")
def set_msg(item: Item):
"""メッセージの登録."""
settings.message = item.message
return {"message": item.message}
# 定義したルーティングをアプリに登録する
app.include_router(router)
ちょっと解説
基本はコード内のコメントに記載しています
ポイントは APIRoute を継承したクラスを定義し get_route_handler をオーバライドする点です
その返り値に自分で定義したハンドラメソッドをリターンするだけです
ハンドラオブジェクトからリクエストからレスポンスを取得、操作できるのでそれを元にロギングやカスタマイズを行います
カスタムルートを元にアプリを作成します
作成したアプリからルーティング定義用のオブジェクトを作成してルーティングを定義します
最後に定義したルーティング情報をアプリに登録すればカスタムルートが適用されたアプリが起動します
0 件のコメント:
コメントを投稿