How to log raw HTTP request/response in Python FastAPI

You may try to customize APIRouter like in FastAPI official documentation:

import time
from typing import Callable

from fastapi import APIRouter, FastAPI, Request, Response
from fastapi.routing import APIRoute


class TimedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            response.headers["X-Response-Time"] = str(duration)
            print(f"route duration: {duration}")
            print(f"route response: {response}")
            print(f"route response headers: {response.headers}")
            return response

        return custom_route_handler


app = FastAPI()
router = APIRouter(route_class=TimedRoute)


@app.get("/")
async def not_timed():
    return {"message": "Not timed"}


@router.get("/timed")
async def timed():
    return {"message": "It's the time of my life"}


app.include_router(router)

You could use a Middleware. A middleware takes each request that comes to your application, and hence, allows you to handle the request before it is processed by any specific endpoint, as well as the response, before it is returned to the client. To create a middleware, you use the decorator @app.middleware("http") on top of a function, as shown below. As you need to consume the request body from the stream inside the middleware—using either request.body() or request.stream(), as shown in this answer (behind the scenes, the former method actually calls the latter, see here)—then it won't be available when you later pass the request to the corresponding endpoint. Thus, you can follow the approach described in this post to make the request body available down the line (i.e., using the set_body function below). As for the response body, you can use the same approach as described in this answer to consume the body and then return the response to the client. Either option described in the aforementioned linked answer would work; the below, however, uses Option 2, which stores the body in a bytes object and returns a custom Response directly (along with the status_code, headers and media_type of the original response).

To log the data, you could use a BackgroundTask, as described in this answer and this answer. A BackgroundTask will run only once the response has been sent (see Starlette documentation as well); thus, the client won't have to be waiting for the logging to complete before receiving the response (and hence, the response time won't be noticeably impacted).

Note: If you had a streaming request or response with a body that wouldn't fit into your server's RAM (for example, imagine a body of 100GB on a machine running 8GB RAM), it would become problematic, as your are storing the data in the RAM, which wouldn't have enough space available to accommodate the accumulated data. You mentioned that "the body size of both request and response JSON is about 1MB"; hence, that should normally be fine (however, it is always a good practice to consider beforehand matters, such as how many requests your API is expected to be serving concurrently, what other applications might be using the RAM, etc., in order to rule whether this is an issue or not). If you needed to, you could limit the number of requests to your API endpoints using, for example, SlowAPI (as shown in this answer). Also, you could limit the usage of the middleware to specific endpoints by checking the request.url.path against a pre-defined list of routes, as described in this answer (see "Update" section), or by using a Custom APIRoute class in a router instead (with similar logic inside as in the middleware below).

Working example

from fastapi import FastAPI, Request, Response
from starlette.background import BackgroundTask
from starlette.types import Message
import logging

app = FastAPI()
logging.basicConfig(filename='example.log', level=logging.DEBUG)
    
def log_req_resp(request_body, response_body):
    logging.info(request_body.decode())
    logging.info(response_body.decode())

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):
    request_body = await request.body()
    await set_body(request, request_body)
    response = await call_next(request)
    
    response_body = b''
    async for chunk in response.body_iterator:
        response_body += chunk
    
    task = BackgroundTask(log_req_resp, request_body, response_body)
    return Response(content=response_body, status_code=response.status_code, 
        headers=dict(response.headers), media_type=response.media_type, background=task)

In case you would like to perform some validation on the request body—for example, ensruing that the request body size is not exceeding a certain value—instead of using request.body(), you can process the body one chunk at a time using the .stream() method, as shown below (similar to this answer).

@app.middleware("http")
async def some_middleware(request: Request, call_next):    
    request_body = b''
    async for chunk in request.stream():
        request_body += chunk
    ...

As the other answers did not work for me and I searched quite extensively on stackoverflow to fix this problem, I will show my solution below.

The main issue is that when using the request body or response body many of the approaches/solutions offered online do simply not work as the request/response body is consumed in reading it from the stream.

To solve this issue I adapted an approach that basically reconstructs the request and response after reading them. This is heavily based on the comment by user 'kovalevvlad' on https://github.com/encode/starlette/issues/495.

Custom middleware is created that is later added to the app to log all requests and responses. Note that you need some kind of logger to store your logs.

from json import JSONDecodeError
import json
import logging
from typing import Callable, Awaitable, Tuple, Dict, List

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response, StreamingResponse
from starlette.types import Scope, Message

# Set up your custom logger here
logger = ""

class RequestWithBody(Request):
    """Creation of new request with body"""
    def __init__(self, scope: Scope, body: bytes) -> None:
        super().__init__(scope, self._receive)
        self._body = body
        self._body_returned = False

    async def _receive(self) -> Message:
        if self._body_returned:
            return {"type": "http.disconnect"}
        else:
            self._body_returned = True
            return {"type": "http.request", "body": self._body, "more_body": False}


class CustomLoggingMiddleware(BaseHTTPMiddleware):
    """
    Use of custom middleware since reading the request body and the response consumes the bytestream.
    Hence this approach to basically generate a new request/response when we read the attributes for logging.
    """
    async def dispatch(  # type: ignore
        self, request: Request, call_next: Callable[[Request], Awaitable[StreamingResponse]]
    ) -> Response:
            # Store request body in a variable and generate new request as it is consumed.
            request_body_bytes = await request.body()
        
        request_with_body = RequestWithBody(request.scope, request_body_bytes)

        # Store response body in a variable and generate new response as it is consumed.
        response = await call_next(request_with_body)
        response_content_bytes, response_headers, response_status = await self._get_response_params(response)

        # Logging

        # If there is no request body handle exception, otherwise convert bytes to JSON.
        try:
            req_body = json.loads(request_body_bytes)
        except JSONDecodeError:
            req_body = ""
        # Logging of relevant variables.
        logger.info(
            f"{request.method} request to {request.url} metadata\n"
            f"\tStatus_code: {response.status_code}\n"
            f"\tRequest_Body: {req_body}\n"
        )
        # Finally, return the newly instantiated response values
        return Response(response_content_bytes, response_status, response_headers)

async def _get_response_params(self, response: StreamingResponse) -> Tuple[bytes, Dict[str, str], int]:
    """Getting the response parameters of a response and create a new response."""
    response_byte_chunks: List[bytes] = []
    response_status: List[int] = []
    response_headers: List[Dict[str, str]] = []

    async def send(message: Message) -> None:
        if message["type"] == "http.response.start":
            response_status.append(message["status"])
            response_headers.append({k.decode("utf8"): v.decode("utf8") for k, v in message["headers"]})
        else:
            response_byte_chunks.append(message["body"])

    await response.stream_response(send)
    content = b"".join(response_byte_chunks)
    return content, response_headers[0], response_status[0]