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

Python FastAPI で生の HTTP リクエストとレスポンスを記録する方法

概要:

Python FastAPI ベースの Web サービスの監査要件を満たすには、生の JSON を保存する必要があります。特定のルート上のリクエストとレスポンスの両方の本文。このガイドでは、約 1MB の本体サイズを扱う場合でも、応答時間に顕著な影響を与えることなくこれを達成するための 2 つの実行可能なソリューションを紹介します。

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

ミドルウェアの仕組み:

ミドルウェアはゲートキーパーとして機能しますアプリケーションに入力されるリクエスト用。これにより、エンドポイント処理の前にリクエストを処理し、クライアントに返す前に応答を処理できるようになります。関数で @app.middleware デコレータを使用してミドルウェアを確立できます:

リクエストおよびレスポンス本文の管理:

ミドルウェア内のストリームからリクエスト本文にアクセスするには ( request.body() または request.stream()) を使用する場合は、リクエストとレスポンスのサイクルの後半で使用できるようにする必要があります。リンクされた投稿では、この回避策について説明していますが、FastAPI バージョン 0.108.0 以降では不要になりました。

応答本文については、この投稿で概要を説明した手法を複製して、本文を直接消費して返し、ステータスを提供できます。コード、ヘッダー、メディア タイプと元の応答。

ロギングデータ:

BackgroundTask を使用してデータをログに記録し、応答完了後の実行を保証します。これにより、クライアントのログ記録タスクの待機がなくなり、応答時間の整合性が維持されます。

オプション 2: カスタム APIRoute の実装

カスタム APIRoute:

このオプションには、処理前にリクエストとレスポンスの本文を操作するためのカスタム APIRoute クラスの作成が含まれますエンドポイント、またはクライアントに結果を返す。専用の APIRouter:

考慮事項:

メモリ制約:

どちらのアプローチでも、使用可能なサーバー RAM を超える大規模なリクエストまたはレスポンスボディに関する問題が発生する可能性があります。大きな応答をストリーミングすると、クライアント側の遅延やリバース プロキシ エラーが発生する可能性があります。潜在的な問題を回避するには、ミドルウェアの使用を特定のルートに制限するか、大量のストリーミング応答を持つエンドポイントを除外します。

コード例 (オプション 1):

from fastapi import FastAPI, APIRouter, Response, Request
from starlette.background import BackgroundTask
from fastapi.routing import APIRoute
from starlette.types import Message
from typing import Dict, Any
import logging


app = FastAPI()
logging.basicConfig(filename='info.log', level=logging.DEBUG)


def log_info(req_body, res_body):
    logging.info(req_body)
    logging.info(res_body)



# Not required for FastAPI >= 0.108.0
async def set_body(request: Request, body: bytes):
    async def receive() -> Message:
        return {'type': 'http.request', 'body': body}
    request._receive = receive


@app.middleware('http')
async def some_middleware(request: Request, call_next):
    req_body = await request.body()
    await set_body(request, req_body)  # Not required for FastAPI >= 0.108.0
    response = await call_next(request)
    
    res_body = b''
    async for chunk in response.body_iterator:
        res_body += chunk
    
    task = BackgroundTask(log_info, req_body, res_body)
    return Response(content=res_body, status_code=response.status_code, 
        headers=dict(response.headers), media_type=response.media_type, background=task)


@app.post('/')
def main(payload: Dict[Any, Any]):
    return payload

例コード (オプション 2):

from fastapi import FastAPI, APIRouter, Response, Request
from starlette.background import BackgroundTask
from starlette.responses import StreamingResponse
from fastapi.routing import APIRoute
from starlette.types import Message
from typing import Callable, Dict, Any
import logging
import httpx


def log_info(req_body, res_body):
    logging.info(req_body)
    logging.info(res_body)

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

        async def custom_route_handler(request: Request) -> Response:
            req_body = await request.body()
            response = await original_route_handler(request)
            tasks = response.background
            
            if isinstance(response, StreamingResponse):
                res_body = b''
                async for item in response.body_iterator:
                    res_body += item
                  
                task = BackgroundTask(log_info, req_body, res_body)
                response = Response(content=res_body, status_code=response.status_code, 
                        headers=dict(response.headers), media_type=response.media_type)
            else:
                task = BackgroundTask(log_info, req_body, response.body)
            
            # Check if the original response had background tasks already attached to it
            if tasks:
                tasks.add_task(task)  # Add the new task to the tasks list
                response.background = tasks
            else:
                response.background = task
                
            return response
            
        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=LoggingRoute)
logging.basicConfig(filename='info.log', level=logging.DEBUG)


@router.post('/')
def main(payload: Dict[Any, Any]):
    return payload


@router.get('/video')
def get_video():
    url = 'https://storage.googleapis.com/gtv-videos-bucket/sample/ForBiggerBlazes.mp4'
    
    def gen():
        with httpx.stream('GET', url) as r:
            for chunk in r.iter_raw():
                yield chunk

    return StreamingResponse(gen(), media_type='video/mp4')


app.include_router(router)

これらのソリューションは以下を提供します応答時間に大きな影響を与えることなく、生の HTTP リクエストとレスポンスの本文を FastAPI に記録するための効率的な方法です。

以上がFastAPI で生の HTTP リクエストとレスポンスの本文を効率的に記録するにはどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。