FastAPI で生の HTTP リクエスト/レスポンスボディを効率的に記録するにはどうすればよいですか?-Python チュートリアル-php.cn

FastAPI で生の HTTP リクエスト/レスポンスボディを効率的に記録するにはどうすればよいですか?

Mary-Kate Olsen

リリース： 2024-12-04 06:43:17

オリジナル

914 人が閲覧しました

How to Efficiently Log Raw HTTP Request/Response Bodies in FastAPI?

Python FastAPI での生の HTTP リクエスト/レスポンスのログ記録

要件:

特定の生の JSON 本文をキャプチャして保存する応答に大きな影響を与えることなく、データサイズが 1MB 程度のリクエストと応答をルーティングします。

オプション 1: ミドルウェアの利用

ミドルウェアの概念

ミドルウェアは、すべてのリクエストがエンドポイントに到達する前にインターセプトし、応答がクライアントに送信される前にインターセプトするため、データ操作が可能になります。ただし、ミドルウェアでリクエスト本文ストリームを使用する場合の問題は、ダウンストリームエンドポイントで利用できなくなることです。したがって、set_body() 関数を使用して利用可能にします。

応答の場合は、BackgroundTask を使用します

ロギングは、BackgroundTask を使用して実行できます。これにより、応答が完了した後にロギングが確実に行われます。クライアントに送信され、応答時間の遅延が回避されます。

ミドルウェア例

# Logging middleware
async def some_middleware(request: Request, call_next):
    req_body = await request.body()
    await set_body(request, req_body)
    response = await call_next(request)

    # Body storage in RAM
    res_body = b''
    async for chunk in response.body_iterator:
        res_body += chunk

    # Background logging task
    task = BackgroundTask(log_info, req_body, res_body)
    return Response(...)

# Endpoint using middleware
@app.post('/')
async def main(payload: Dict):
    pass

ログイン後にコピー

オプション 2: カスタム APIRoute クラス

APIRoute クラス拡張

カスタム APIRoute クラスを作成することで、リクエストとレスポンスの本文を制御し、その使用を次のように制限できます。 APIRouter を介した特定のルート。

重要考慮事項

大規模な応答 (ストリーミングメディアなど) の場合、カスタムルートでは、応答全体を RAM に読み取るために RAM の問題やクライアント側の遅延が発生する可能性があります。したがって、そのようなエンドポイントをカスタムルートから除外することを検討してください。

カスタム APIRoute クラスの例

class LoggingRoute(APIRoute):
    async def custom_route_handler(request: Request) -> Response:
        req_body = await request.body()
        response = await original_route_handler(request)

        # Response handling based on type
        if isinstance(response, StreamingResponse):
            res_body = b''
            async for item in response.body_iterator:
                res_body += item
            response = Response(...)
        else:
            response.body

        # Logging task
        task = BackgroundTask(log_info, req_body, response.body)
        response.background = task
        return response

# Endpoint using custom APIRoute
@router.post('/')
async def main(payload: Dict):
    return payload

ログイン後にコピー