I have a FastAPI app running over uvicorn. I would like to achieve the following goals:
- Catch uvicorn logs and process them with Loguru. (Done with InterceptHandler)
- Get request and response body and log in debug mode. (Done with APIRoute)
- Log into two files: log.log where one line = one logging message, no traceback. The second file - traceback.log = error logging message + traceback. (Not solved, need help)
What I've tried:
- https://github.com/Delgan/loguru/issues/103 - the problem with this solution is that it doesn't work automatically. Currently, I get traceback from anywhere
- Python logging: disable stack trace - looks like the right thing, but I don't understand, how to add to my code
- Python - loguru disable traceback on exceptions - not solving the problem
My code:
main.py:
from fastapi import FastAPI
from rest_api.logging import init_logging, BodyLoggingRoute
app = FastAPI()
app.add_event_handler("startup", init_logging) # to redirect all unicorn logs to loguru log file
app.router.route_class = BodyLoggingRoute
rest_api/logging.py:
import logging
from pathlib import Path
from typing import Callable
import json
from loguru import logger
from fastapi.routing import APIRoute
from fastapi import Request, Response
class InterceptHandler(logging.Handler):
"""
Default handler from examples in loguru documentaion.
See https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging
"""
def emit(self, record: logging.LogRecord):
# Get corresponding Loguru level if it exists
try:
level = logger.level(record.levelname).name
except ValueError:
level = record.levelno
# Find caller from where originated the logged message
frame, depth = logging.currentframe(), 2
while frame.f_code.co_filename == logging.__file__:
frame = frame.f_back
depth += 1
logger.opt(depth=depth, exception=record.exc_info).log(
level, record.getMessage()
)
def init_logging(min_level='INFO'):
"""
Replaces logging handlers with a handler for using the custom handler.
"""
def level_filter(record):
return record["level"].no >= logger.level(min_level).no
# # change handler for default uvicorn logger
intercept_handler = InterceptHandler()
logging.getLogger("uvicorn.access").handlers = [intercept_handler]
logging.getLogger("uvicorn").handlers = [intercept_handler]
log_path = Path(__file__).parents[2].resolve() / 'logging'
logger.add(str(log_path / 'log.log'), format='{time} {level} {message}',
filter=level_filter, rotation='1 month', compression='zip')
class BodyLoggingRoute(APIRoute):
"""
https://fastapi.tiangolo.com/advanced/custom-request-and-route/
"""
def get_route_handler(self) -> Callable:
original_route_handler = super().get_route_handler()
async def body_logging_route_handler(request: Request) -> Response:
req_body = await request.body()
response = await original_route_handler(request)
res_body = response.body
req_body = json.loads(req_body) if req_body else {}
logger.debug(f'Request Body: {req_body}')
res_body = json.loads(res_body) if res_body else {}
logger.debug(f'Response JSON: {res_body}')
return response
return body_logging_route_handler