FastAPI에서 원시 HTTP 요청/응답 본문을 효율적으로 기록하는 방법은 무엇입니까?-파이썬 튜토리얼-php.cn

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

Mary-Kate Olsen

풀어 주다： 2024-12-04 06:43:17

원래의

919명이 탐색했습니다.

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

로그인 후 복사