FastAPI log every request and response

loggedroute.py

from fastapi import Request, Response, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.routing import APIRoute
from http.client import responses
from json import JSONDecodeError
from starlette.exceptions import HTTPException as StarletteHTTPException
from traceback import format_exc
from typing import Callable
import logging

# Python builtin logging
logging.basicConfig(
    format="%(asctime)s [%(process)d] [%(levelname)s] %(message)s",
    level=logging.INFO
)

# FastAPI custom logging
class LoggedRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original_route_handler = super().get_route_handler()

        async def custom_route_handler(request: Request) -> Response:
            request_params = ""

            try:
                # Decode request JSON body
                request_json = await request.json()
                request_params += " ".join(f"{k}={v}" for k, v in request_json.items())
            except JSONDecodeError:
                # Request has no JSON body
                pass

            # Decode request query params
            if request.query_params:
                request_params += " " if request_params else "" # add space for pretty printing
                request_params += " ".join(f"{k}={v}" for k, v in request.query_params.items())

            # Log the request
            route_called = f"{request.method} {request.url.path} ({request_params})"
            request_log = f"Request from {request.client.host}:{request.client.port}: {route_called}"
            response_log = f"Response to {request.client.host}:{request.client.port}: {route_called}"
            logging.info(request_log)

            # Log the response
            try:
                response: Response = await original_route_handler(request)

                # Successful response
                success_info = f"{response.status_code} {responses[response.status_code]}"
                logging.info(f"{response_log} - {success_info}")
                return response
            except (HTTPException, StarletteHTTPException) as exc:
                # HTTP exception
                error_info = f"{exc.status_code} {responses[exc.status_code]}: {exc.detail}"
                logging.info(f"{response_log} - {error_info}")
                raise exc
            except RequestValidationError as exc:
                # Validation error
                error_info = f"422 {responses[422]}: {exc.__class__.__name__}"
                logging.info(f"{response_log} - {error_info}")
                raise exc
            except Exception as exc:
                # Any other error
                error_info = f"500 {responses[500]}: {exc.__class__.__name__}"
                error_trace = format_exc()
                logging.info(f"{response_log} - {error_info}\n{error_trace}")
                raise HTTPException(500, error_trace)

        return custom_route_handler

What for?

This snippet helps to log every route call. The log will include:

• request parameters (JSON body and query parameters)
• response status code (and corresponding name) with a detail message in case of an HTTP error

It uses the APIRoute advanced feature of FastAPI (as per Custom Request and APIRoute class). In particular, it explores both Request and Response classes from Starlette.

Quick demo

# api.py

from fastapi import FastAPI, HTTPException
from loggedroute import LoggedRoute
from pydantic import BaseModel

app = FastAPI()
app.router.route_class = LoggedRoute


class Item(BaseModel):
    name: str
    price: float

    def __init__(self, **data):
        super().__init__(**data)

        if self.name == "reserved":
            raise ValueError("'name' cannot be 'reserved'")


@app.get("/")
async def read_root():
    return {"Hello": "World"}


@app.post("/items", status_code=201)
def post_item(item: Item, convert_price: bool = False):
    if item.name == "already-taken":
        raise HTTPException(409, "Given item name already exists")
    return item


@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id == 0:
        raise HTTPException(404, "Given item does not exist")
    if item_id == -1:
        print(not_defined_var)  # will trigger an Internal Server Error

Execution log:

$ uvicorn api:app --reload --no-access-log # we disable access log to better visualize our custom logging

INFO:     Will watch for changes in these directories: ['/tmp/fa']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [11507] using StatReload
INFO:     Started server process [11509]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

2024-04-27 13:45:42,401 [11102] [INFO] Request from 127.0.0.1:49848: GET / ()
2024-04-27 13:45:42,401 [11102] [INFO] Response to 127.0.0.1:49848: GET / () - 200 OK

2024-04-27 13:45:57,413 [11102] [INFO] Request from 127.0.0.1:35902: POST /items (name=my-item price=5 convert_price=false)
2024-04-27 13:45:57,417 [11102] [INFO] Response to 127.0.0.1:35902: POST /items (name=my-item price=5 convert_price=false) - 201 Created

2024-04-27 13:46:05,564 [11102] [INFO] Request from 127.0.0.1:35912: POST /items (name=my-other-item price=10 convert_price=true)
2024-04-27 13:46:05,565 [11102] [INFO] Response to 127.0.0.1:35912: POST /items (name=my-other-item price=10 convert_price=true) - 201 Created

2024-04-27 13:46:35,228 [11102] [INFO] Request from 127.0.0.1:46716: POST /items (name=reserved price=0 convert_price=true)
2024-04-27 13:46:35,228 [11102] [INFO] Response to 127.0.0.1:46716: POST /items (name=reserved price=0 convert_price=true) - 422 Unprocessable Entity: RequestValidationError

2024-04-27 13:46:46,970 [11102] [INFO] Request from 127.0.0.1:56064: POST /items (name=already-taken price=0 convert_price=true)
2024-04-27 13:46:46,971 [11102] [INFO] Response to 127.0.0.1:56064: POST /items (name=already-taken price=0 convert_price=true) - 409 Conflict: Given item name already exists

2024-04-27 13:47:03,093 [11102] [INFO] Request from 127.0.0.1:49888: GET /items/0 ()
2024-04-27 13:47:03,093 [11102] [INFO] Response to 127.0.0.1:49888: GET /items/0 () - 404 Not Found: Given item does not exist

2024-04-27 13:47:09,972 [11102] [INFO] Request from 127.0.0.1:43408: GET /items/-1 ()
2024-04-27 13:47:09,978 [11102] [INFO] Response to 127.0.0.1:43408: GET /items/-1 () - 500 Internal Server Error: NameError
Traceback (most recent call last):
  File "/tmp/sample/loggedroute.py", line 46, in custom_route_handler
    response: Response = await original_route_handler(request)
  File "/tmp/sample/venv/lib/python3.10/site-packages/fastapi/routing.py", line 278, in app
    raw_response = await run_endpoint_function(
  File "/tmp/sample/venv/lib/python3.10/site-packages/fastapi/routing.py", line 193, in run_endpoint_function
    return await run_in_threadpool(dependant.call, **values)
  File "/tmp/sample/venv/lib/python3.10/site-packages/starlette/concurrency.py", line 42, in run_in_threadpool
    return await anyio.to_thread.run_sync(func, *args)
  File "/tmp/sample/venv/lib/python3.10/site-packages/anyio/to_thread.py", line 56, in run_sync
    return await get_async_backend().run_sync_in_worker_thread(
  File "/tmp/sample/venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 2144, in run_sync_in_worker_thread
    return await future
  File "/tmp/sample/venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 851, in run
    result = context.run(func, *args)
  File "/tmp/sample/api.py", line 39, in get_item
    print(not_defined_var)  # will trigger an Internal Server Error
NameError: name 'not_defined_var' is not defined

2024-04-27 13:47:17,623 [11102] [INFO] Request from 127.0.0.1:49064: GET /items/1 ()
2024-04-27 13:47:17,623 [11102] [INFO] Response to 127.0.0.1:49064: GET /items/1 () - 200 OK

Note the error trace is logged (and returned to the client as well) so that it helps for debugging.

Notes

1. For bigger applications, you can set the route_class at the router level (as per Custom APIRoute class in a router):

app = FastAPI()
router = APIRouter(route_class=LoggedRoute)
app.include_router(router)

2. As per FastAPI's HTTPException vs Starlette's HTTPException, it is advised to catch both HTTPException and StarletteHTTPException:

This way, if any part of Starlette's internal code, or a Starlette extension or plug-in, raises a Starlette HTTPException, your handler will be able to catch and handle it.