How to Efficiently Log Raw HTTP Request and Response Bodies in FastAPI?

How to Record Raw HTTP Requests and Responses in Python FastAPI

Introduction:

In order to meet auditing requirements for your Python FastAPI-based web service, you need to preserve the raw JSON bodies of both requests and responses on certain routes. This guide will present two viable solutions to accomplish this without noticeably impacting response times, even when working with body sizes approximating 1MB.

Option 1: Middleware Utilization

Middleware Mechanics:

Middleware functions as a gatekeeper for requests entering the application. It allows for handling requests before endpoint processing and responses before returning to clients. You can establish middleware using the @app.middleware decorator on a function:

Request and Response Body Management:

To access the request body from the stream within middleware (using request.body() or request.stream()), you'll need to make it available later in the request-response cycle. The linked post discusses this workaround, which now is unnecessary for FastAPI versions 0.108.0 and above.

For the response body, you can replicate the technique outlined in this post to consume and return the body directly, providing status code, headers, and media type along with the original response.

Logging Data:

Employ BackgroundTask to log data, ensuring its execution after response completion. This eliminates client waiting for logging tasks and maintains response time integrity.

Option 2: Custom APIRoute Implementation

Custom APIRoute:

This option involves creating a custom APIRoute class for manipulating request and response bodies before processing endpoints or returning results to clients. It enables the isolation of custom route handling to specific endpoints by using a dedicated APIRouter:

Considerations:

Memory Constraints:

Both approaches may encounter challenges with large request or response bodies exceeding available server RAM. Streaming large responses can introduce client-side delays or reverse proxy errors. Restrict middleware usage to specific routes or exclude endpoints with large streaming responses to avoid potential issues.

Example Code (Option 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

Example Code (Option 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)

These solutions provide efficient methods for logging raw HTTP request and response bodies in FastAPI without significantly affecting response times.

The above is the detailed content of How to Efficiently Log Raw HTTP Request and Response Bodies in FastAPI?. For more information, please follow other related articles on the PHP Chinese website!