💨
【FastAPI】リクエストとレスポンスのログを出力する
Python の logger モジュールを使ってリクエスト情報とレスポンス情報をログに出力する方法をメモとして残しておく
APIRoute クラスを使うことで実現できるらしいので調べてみる
Custom Request and APIRoute class - FastAPI
ユースケース
・すべてのリクエスト本文を自動的にログに記録します。
APIRoute クラスとは？
APIRoute は、APIRouter のコンストラクタの route_class に指定できるもの Docs
これのクラスを継承したクラスを使うことでリクエスト処理の前後に処理を挟むことができる
CustomAPIRoute クラスを指定する場合は以下のようになる
from fastapi import APIRouter
router = APIRouter(route_class=CustomAPIRoute)
@router.get("/items/")
async def read_items():
return {"message": "Item list"}
APIRouter とは？
APIRouter は、ルートをグループ化できるようにすることで、モジュールの分割をできるようにする仕組み
この機能を使うことで各機能やリソースごとにコードを整理できるようになる
APIRouter を使ったルートの定義
from fastapi import APIRouter
router = APIRouter()
@router.get("/items/")
async def read_items():
return {"message": "Item list"}
アプリケーションへの組み込み
from fastapi import FastAPI
app = FastAPI()
app.include_router(router)
カスタマイズした APIRoute を使ってリクエスト処理の前後に処理を挟む
Docs にあるように、fastapi.routing.APIRoute のサブクラスを作成して、APIRoute.get_route_handler() をオーバーライドすることによって実現できる
APIRoute.get_route_handler() は関数を返す。この関数によってリクエストの処理が行われるようになっている。そのため、super().get_route_handler() を保持しておき、それをサブクラス内の get_route_handler() で返す関数の中で呼び出すことで処理を挟むことができる
通常のAPIRouteと同じことができる最小のカスタムルーターは以下のようになる
from typing import Any, Callable, Coroutine
from fastapi import Request, Response
from fastapi.routing import APIRoute
class LoggingAPIRoute(APIRoute):
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
# もとのリクエスト処理ハンドラを保持しておく (この関数でリクエストを処理する)
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
# ここに処理を挟める
response = await original_route_handler(request)
# ここに処理を挟める
return response
return custom_route_handler
これを書くAPIRouterに渡してあげる
router = APIRouter(route_class=LoggingAPIRoute)
ログ出力する APIRoute
いろいろ試しながら以下のような APIRoute ができた
_get_request_body の実装は、FastAPI の APIRoute の実装を参考にして書いた
import email.message
import json
import time
from logging import getLogger
from typing import Any, Callable, Coroutine
from fastapi import Request, Response, params
from fastapi.routing import APIRoute
logger = getLogger(f"uvicorn.{__name__}")
class LoggingAPIRoute(APIRoute):
"""リクエストとレスポンス情報をログ出力するAPIRoute"""
def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
# もとのリクエスト処理ハンドラを保持しておく (この関数でリクエストを処理する)
original_route_handler = super().get_route_handler()
async def custom_route_handler(request: Request) -> Response:
before = time.time()
await self._logging_request_body(request)
# リクエスト処理を実行
response = await original_route_handler(request)
duration_sec = time.time() - before
self._logging_response_body(response, request, duration_sec)
return response
return custom_route_handler
async def _logging_request_body(self, request: Request):
body = await self._get_request_body(request)
request_info = {
"method": request.method,
"url": str(request.url),
"body": body,
}
logger.info("start request", extra={"request_info": request_info})
def _logging_response_body(
self, response: Response, request: Request, duration_sec: float
):
body = self._get_response_body(response)
response_info = {
"method": request.method,
"url": str(request.url),
"status_code": response.status_code,
"body": body,
"headers": response.headers,
"duration_sec": round(duration_sec, 5),
}
logger.info("end request", extra={"response_info": response_info})
async def _get_request_body(self, request: Request):
"""fastapi/routing.py の app 関数のように body の内容を判定して出力する"""
is_body_form = self.body_field and isinstance(
self.body_field.field_info, params.Form
)
if is_body_form:
return await request.form()
body_bytes = await request.body()
if not body_bytes:
return None
content_type_value = request.headers.get("content-type")
if not content_type_value:
return await request.json()
if self._is_content_type_json(content_type_value):
return await request.json()
return body_bytes.decode("utf-8")
def _get_response_body(self, response: Response):
if response.body == b"":
return None
content_type_value = response.headers.get("content-type")
if not content_type_value:
# XXX: JSON とみなしていいか不明なため、そのまま返している
return response.body
# return json.loads(response.body)
if self._is_content_type_json(content_type_value):
return json.loads(response.body)
# TODO: 他にも考慮が必要なパターンがあれば、追加する
return response.body.decode("utf-8")
def _is_content_type_json(self, content_type_value: str):
message = email.message.Message()
message["content-type"] = content_type_value
if message.get_content_maintype() != "application":
return False
subtype = message.get_content_subtype()
return subtype == "json" or subtype.endswith("+json")
Discussion