FastAPI에서 원시 HTTP 요청 및 응답 본문을 효율적으로 기록하는 방법은 무엇입니까?

Python FastAPI에서 원시 HTTP 요청 및 응답을 기록하는 방법

소개:

In Python FastAPI 기반 웹 서비스에 대한 감사 요구 사항을 충족하려면 다음의 원시 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 중국어 웹사이트의 기타 관련 기사를 참조하세요!