2

I have a FastAPI app running over uvicorn. I would like to achieve the following goals:

  1. Catch uvicorn logs and process them with Loguru. (Done with InterceptHandler)
  2. Get request and response body and log in debug mode. (Done with APIRoute)
  3. 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:

  1. 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
  2. Python logging: disable stack trace - looks like the right thing, but I don't understand, how to add to my code
  3. 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

Ivan Mishalkin
  • 1,049
  • 9
  • 25

0 Answers0