Logging request and response in your FastAPI app.

I have encountered this few times in the past. If you search on the Internet, you may see some approaches that deal with this requirement by custom middleware.

For example, https://medium.com/@dhavalsavalia/fastapi-logging-middleware-logging-requests-and-responses-with-ease-and-style-201b9aa4001a

https://github.com/tiangolo/fastapi/discussions/8187

https://github.com/encode/starlette/issues/495

After looking into multiple Github issues from FastAPI and Starlette, I think custom route should be the better way to go.

https://fastapi.tiangolo.com/how-to/custom-request-and-route

In this approach, you don’t need to deal with hanging forever issue after calling await request.body() in dispatch.

The following code snippet is my implementation.

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

        async def custom_route_handler(request: Request) -> Response:
            host = getattr(getattr(request, "client", None), "host", None)
            port = getattr(getattr(request, "client", None), "port", None)

            url = f"{request.url.path}?{request.query_params}" if request.query_params else request.url.path

            headers = dict(request.scope['headers'])

            request_id = headers.get(b"x-request-id")

            if not request_id:
                request_id = str(uuid.uuid4())
                headers[b"x-request-id"] = bytes(request_id, "utf-8")
            else:
                request_id = str(request_id, 'UTF-8')

            request.scope['headers'] = [(k, v) for k, v in headers.items()]

            request_body = await request.body()
            before = time.time()
            response: Response = await original_route_handler(request)
            duration = time.time() - before
            duration = format(duration, '.2f')
            response.headers["X-Process-Time"] = duration
            response.headers["X-Request-Id"] = request_id

            try:
                request_body = json.loads(request_body)
            except JSONDecodeError:
                request_body = {}
            except UnicodeDecodeError:
                request_body = {}

            try:
                response_body = json.loads(response.body)
            except JSONDecodeError:
                response_body = {}

            try:
                status_phrase = http.HTTPStatus(response.status_code).phrase
            except ValueError:
                status_phrase = ""

            logging.info(
                f'{request_id} - '
                f'{host}:{port} - "{request.method} {url}" '
                f'{response.status_code} {status_phrase} {str(duration)}s '
                f'request_header: {dict(request.headers)} '
                f'request_body: {request_body} '
                f'response_header: {dict(response.headers)} '
                f'response_body: {response_body} '
            )
            return response

        return custom_route_handler