如何在 FastAPI 中有效記錄原始 HTTP 請求和回應正文？

如何記錄原始HTTP為了滿足基於Python FastAPI 的Web 服務的審核要求，您需要保留以下原始JSON主體某些路由上的請求和回應。本指南將提供兩種可行的解決方案來實現此目的，即使在處理大小約為 1MB 的正文時，也不會明顯影響回應時間。

選項1：中間件利用

中間件機制：

中間件請求的看門人進入應用程式。它允許在端點處理之前處理請求，並在返回客戶端之前處理回應。您可以在函數上使用@app.middleware 裝飾器來建立中間件：

請求和回應正文管理：

從中間件中的流存取請求正文（使用request.body() 或request.stream())，您需要稍後在請求-回應週期中使其可用。連結的貼文討論了此解決方法，現在對於 FastAPI 版本 0.108.0 及更高版本來說，這是不必要的。

對於回應正文，您可以複製本文中概述的技術來直接使用並返回正文，提供狀態代碼、標頭和媒體類型以及原始回應。

日誌記錄資料：

使用BackgroundTask來記錄數據，確保回應完成後執行。這消除了客戶端等待日誌任務並保持回應時間完整性。

選項2：自訂APIRoute 實作

自訂APIRoute:

此選項涉及建立一個自訂APIRoute 類別，用於在處理端點或將結果傳回給客戶端之前操作請求和回應主體。它可以透過使用專用APIRouter 將自訂路由處理隔離到特定端點：

注意事項：

記憶體約束：

記憶體約束：

記憶體約束：

記憶體約束：

兩種方法都可能會遇到超過可用伺服器RAM 的大型請求或回應主體的挑戰。串流傳輸大型回應可能會導致客戶端延遲或反向代理錯誤。將中介軟體使用限制為特定路由或排除具有大量串流響應的端點，以避免潛在問題。

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

範例程式碼（選項1）：

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)

範例程式碼（選項2）：這些解決方案提供了高效能在FastAPI 記錄原始HTTP 請求和回應主體而不顯著影響回應時間的方法。

以上是如何在 FastAPI 中有效記錄原始 HTTP 請求和回應正文？的詳細內容。更多資訊請關注PHP中文網其他相關文章！