FastAPI Logging
Because of the Starlette ASGI server, it is more difficult to incorporate the Fastapi logging system in the middleware.
In starlette, the request body is an async stream. In order to log the request body, this stream must be consumed. The request body can be consumed in a several ways, such as request.body() and request.stream().
Check out the below example of starlette is used to read the body request object.
from starlette.requests import Request
from starlette.responses import Response
async def app(scope, receive, send):
assert scope['type'] == 'http'
request = Request(scope, receive)
body = b''
async for chunk in request.stream():
body += chunk
response = Response(body, media_type='text/plain')
await response(scope, receive, send)
If the request body is consumed in the middleware it’s impossible for any subsequent methods to access it, there are some workaround to do it in middleware itself.
Log using the router
Alternatively we can use the router to add the logs. Overriding the root router and inheriting it in all the API router as a parent class will does the work. One more advantage with this logging can be done to specific endpoints rather than allowing to all in middleware.
This is two steps process
- Create custom APIrouter class with logging
from typing import Callable
from fastapi import Request, Response
from fastapi.routing import APIRoute
logger = logging.getLogger(__name__)
class CustomLogRouter(APIRoute):
"""
Logs every request and response payload.
NOTE: Explicitly need to add the route class in every new routes else this doesn't log the request and response. E.G., router = APIRouter(route_class=CustomLogRouter)
"""
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def request_response_logger(request: Request) -> Response:
logger.info(f"Request Path: {request.method} {request.url}")
logger.info(f"Request query params:{request.query_params}")
logger.info(f"Request path params:{request.path_params}")
if "content-type" in request.headers:
if request.headers["content-type"].startswith("multipart/form-data;"):
logger.info(f"Request Form body: {await request.form()}")
else:
logger.info(f"Request JSON body: {await request.body()}")
else:
pass
try:
response: Response = await original_route_handler(request)
except Exception as e:
logger.error(f"Unknown error: {e}")
raise HTTPException(500)
else:
if response.media_type == "application/json":
logger.info(f"Response Body: {response.body}")
else:
logger.info(
f"Response Body media type: {response.media_type}, head: {response.headers}"
)
return response
return request_response_logger
else:
if response.media_type == "application/json":
logger.info(f"Response Body: {response.body}")
else:
logger.info(
f"Response Body media type: {response.media_type}, head: {response.headers}"
)
return response
return request_response_logger
- Inherit Root Router to router class
from fastapi import APIRouter
router = APIRouter(route_class=CustomLogRouter)
Reference Links:
-
Logging incoming request and response = github issue on logging
-
Logging with middleware and router example = stackoverflow
-
Starlette request body consuming = starlette requests