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.
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
搶先發佈留言