FastAPIでリクエストとレスポンスの中身をログ出力する

2021/02/07 追記例外時にエラー内容とスタックトレースを含めてロギングする方法を以下の記事で紹介しています。

FastAPI に触れる機会があって、ロギングについてあれこれ考えています。Learn Microservices Logging Best Practices にあるようにリクエスト毎にコンテキストをロギングした方が良さそうです。さらに詳細に調べたい場合に備えてリクエストボディとレスポンスボディの中身もロギングした方が良い場合もあります。

Nginx などをリバースプロキシとして置くことでロギングはできるかと思いますが、今回は FastAPI のみでロギングできる方法を調べてみました。

FastAPI とは
サンプルアプリ
Middleware を使う方法
カスタムルータを実装する方法
ログ出力の確認
今後

FastAPI とは

FastAPI は Python のウェブフレームワークです。パフォーマンスの高さなどで最近注目されており、API を実装するのに向いています。

fastapi.tiangolo.com

サンプルアプリ

GitHub に置いています。

github.com

API の中身はドキュメントにあるチュートリアルレベルです。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}


@app.post("/items/")
def post_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}


@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

Middleware を使う方法

Middleware はリクエストを受け取ってレスポンスを返す間に処理をすることができます。ドキュメントには処理時間をカスタムヘッダ（X-Process-Time）に入れる例が載っています。

import time

from fastapi import FastAPI, Request

app = FastAPI()


@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

これを応用して、リクエスト毎に ID を振ってログ出力している例が以下の記事で紹介されています。

FastAPI logging. Here is how I setup my python logging… | by Philippe Girard | Medium

ただ、この方法だとリクエストとレスポンスの中身を取り出すことができません（少なくとも私には難しいです）。Stack Overflow を見るとルーターをカスタマイズする方法が紹介されていました。

python - Get starlette request body in the middleware context - Stack Overflow

カスタムルータを実装する方法

ドキュメントによると、アプリケーションで処理される前にリクエストボディを取得したり操作する場合にこのケースが良いとのことです。

ユースケースは以下のとおりです。まさしく3つ目が今回やりたかったことなのでこの方法を採用しました。

msgpack など非 JSON リクエストボディを JSON に変換する
gzip で圧縮されたリクエストボディを解凍する
すべてのリクエストボディをロギングする

以下にスクリプトを載せます。ただし、ドキュメントにある通り、アドバンスドな機能なので、各自責任の上でご利用ください。

import time
import datetime
import sys
import json
from logging import getLogger, StreamHandler, DEBUG
from typing import Callable
from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute
from pydantic import BaseModel


logger = getLogger(__name__)
handler = StreamHandler(sys.stdout)
handler.setLevel(DEBUG)
logger.addHandler(handler)
logger.setLevel(DEBUG)


class LoggingContextRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            """
            時間計測
            """
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = round(time.time() - before, 4)

            record = {}
            time_local = datetime.datetime.fromtimestamp(before)
            record["time_local"] = time_local.strftime("%Y/%m/%d %H:%M:%S%Z")
            if await request.body():
                record["request_body"] = (await request.body()).decode("utf-8")
            record["request_headers"] = {
                k.decode("utf-8"): v.decode("utf-8") for (k, v) in request.headers.raw
            }
            record["remote_addr"] = request.client.host
            record["request_uri"] = request.url.path
            record["request_method"] = request.method
            record["request_time"] = str(duration)
            record["status"] = response.status_code
            record["response_body"] = response.body.decode("utf-8")
            record["response_headers"] = {
                k.decode("utf-8"): v.decode("utf-8") for (k, v) in response.headers.raw
            }
            logger.info(json.dumps(record))
            return response

        return custom_route_handler


app = FastAPI()
app.router.route_class = LoggingContextRoute


class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}


@app.post("/items/")
def post_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}


@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_name": item.name, "item_id": item_id}

APIRoute クラスを継承して get_route_handler() をオーバーライドすることでルータをカスタマイズしています。取得したリクエストボディなどを JSON 形式にしてロギングしています。フォーマット名は Nginx のログ形式を参考にしています。

time_local: リクエストを受け取った日時
request_body: リクエストボディ
request_headers: リクエストヘッダ
remote_addr: アクセス元 IP アドレス
request_uri: リクエストパス
request_method: リクエストメソッド
request_time: リクエストの処理時間（秒）
status: HTTPステータスコード
response_body: レスポンスボディ
response_headers: レスポンスヘッダ

ログ出力の確認

curl でリクエストしてみてみます。

# GET
$ curl localhost:8000/
$ curl localhost:8000/items/12?foo=bar

# POST
$ curl -X POST "http://127.0.0.1:8000/items/" \
  -H "accept: application/json" \
  -H "Content-Type: application/json" \
  -d "{\"name\":\"foo\",\"price\":1234,\"is_offer\":true}"

# PUT
curl -X PUT "http://127.0.0.1:8000/items/5678" \
  -H  "accept: application/json" \
  -H  "Content-Type: application/json" \
  -d "{\"name\":\"foo\",\"price\":100,\"is_offer\":true}"

ログは以下のとおりです。uvicorn のアクセスログも出ていますが、取得したいログが出ていることがわかります。

{"time_local": "2021/01/31 10:29:45", "request_headers": {"host": "localhost:8000", "user-agent": "curl/7.68.0", "accept": "*/*"}, "remote_addr": "127.0.0.1", "request_uri": "/", "request_method": "GET", "request_time": "0.0008", "status": 200, "response_body": "{\"Hello\":\"World\"}", "response_headers": {"content-length": "17", "content-type": "application/json"}}
INFO:     127.0.0.1:37996 - "GET / HTTP/1.1" 200 OK

{"time_local": "2021/01/31 10:29:45", "request_headers": {"host": "localhost:8000", "user-agent": "curl/7.68.0", "accept": "*/*"}, "remote_addr": "127.0.0.1", "request_uri": "/items/12", "request_method": "GET", "request_time": "0.0005", "status": 200, "response_body": "{\"item_id\":12,\"q\":null}", "response_headers": {"content-length": "23", "content-type": "application/json"}}
INFO:     127.0.0.1:37998 - "GET /items/12?foo=bar HTTP/1.1" 200 OK

{"time_local": "2021/01/31 10:29:45", "request_body": "{\"name\":\"foo\",\"price\":1234,\"is_offer\":true}", "request_headers": {"host": "127.0.0.1:8000", "user-agent": "curl/7.68.0", "accept": "application/json", "content-type": "application/json", "content-length": "43"}, "remote_addr": "127.0.0.1", "request_uri": "/items/", "request_method": "POST", "request_time": "0.0005", "status": 200, "response_body": "{\"item_name\":\"foo\",\"item_price\":1234.0}", "response_headers": {"content-length": "39", "content-type": "application/json"}}
INFO:     127.0.0.1:38000 - "POST /items/ HTTP/1.1" 200 OK

{"time_local": "2021/01/31 10:29:45", "request_body": "{\"name\":\"foo\",\"price\":100,\"is_offer\":true}", "request_headers": {"host": "127.0.0.1:8000", "user-agent": "curl/7.68.0", "accept": "application/json", "content-type": "application/json", "content-length": "42"}, "remote_addr": "127.0.0.1", "request_uri": "/items/5678", "request_method": "PUT", "request_time": "0.0004", "status": 200, "response_body": "{\"item_name\":\"foo\",\"item_id\":5678}", "response_headers": {"content-length": "34", "content-type": "application/json"}}
INFO:     127.0.0.1:38002 - "PUT /items/5678 HTTP/1.1" 200 OK

今後

上の方法だけでは例外処理などによるスタックトレースはロギングされない（JSON 形式に出力されない）ので改良の余地はあります。

本日も乙

ただの自己満足な備忘録。