如何在 FastAPI 中高效记录原始 HTTP 请求和响应正文？

如何在 Python FastAPI 中记录原始 HTTP 请求和响应

简介：

中为了满足基于 Python FastAPI 的 Web 服务的审核要求，您需要保留以下原始 JSON 主体某些路由上的请求和响应。本指南将提供两种可行的解决方案来实现此目的，即使在处理大小约为 1MB 的正文时，也不会明显影响响应时间。

选项 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)

这些解决方案提供了高效在 FastAPI 中记录原始 HTTP 请求和响应主体而不显着影响响应时间的方法。

以上是如何在 FastAPI 中高效记录原始 HTTP 请求和响应正文？的详细内容。更多信息请关注PHP中文网其他相关文章！