FastAPI中自定义HTTPException对接自定义日志系统的实现问题

在FastAPI中使用自定义日志系统时，抛出HTTPException(status_code=[状态码], detail=[信息])时，FastAPI默认输出的日志级别为INFO，但实际期望是ERROR。需要实现一个可灵活指定日志级别（如INFO、ERROR、WARNING等）的自定义异常，同时保留返回给用户的detail字段，并且对接现有日志系统。

现有日志文件log_file.py代码：

import logging.handlers
import logging
import datetime
import yaml

class CustomFormatter(logging.Formatter):
    LOG_DIR = 'server_logs/'

    blue = '\x1b[38;5;39m'
    green = '\x1b[1;32m'
    yellow = '\x1b[38;5;226m'
    red = '\x1b[38;5;196m'
    bold_red = '\x1b[31;1m'
    reset = '\x1b[0m'

    data = '%(asctime)s | '
    level_name = '%(levelname)8s '
    file_and_message = '| (%(filename)s:%(lineno)d) | %(message)s'

    FORMATS = {
            logging.DEBUG: data + blue + level_name + reset + file_and_message,
            logging.INFO: data + green + level_name + reset + file_and_message,
            logging.WARNING: data + yellow + level_name + reset + file_and_message,
            logging.ERROR: data + red + level_name + reset + file_and_message,
            logging.CRITICAL: data + bold_red + level_name + reset + file_and_message
        }

    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt)
        return formatter.format(record)

class Logger:
    with open("server_config.yaml", "r+") as f:
        server_config = yaml.safe_load(f)
    logger = logging.getLogger()
    logger.setLevel(server_config['LOG_LEVEL_CONSOLE'].upper())
    logger.handlers = []

    # output custom log format in console
    console = logging.StreamHandler()
    console.setLevel(server_config['LOG_LEVEL'].upper())
    console.setFormatter(CustomFormatter())

    # save custom logs format to file
    today = datetime.date.today()
    save_in_file = logging.handlers.RotatingFileHandler(
        CustomFormatter.LOG_DIR +
        'Server_application{}.log'.format(today.strftime('%Y_%m_%d'))
    )
    save_in_file.setLevel(logging.DEBUG)
    save_in_file.setFormatter(CustomFormatter())

    # Add both handlers to the logger
    logger.addHandler(console)
    logger.addHandler(save_in_file)

抛出HTTPException时的日志示例：

2023-11-05 17:37:31,538 | INFO | (h11_impl.py:478) | 127.0.0.1:54249 - "GET /vehicle/all HTTP/1.1" 404

曾尝试自定义Exception但未达预期，示例代码：

class CustomException(Exception):
    def __init__(self, name: str, info: str):
        self.status_code = name
        self.detail = info


@app.exception_handler(CustomException)
async def unicorn_exception_handler(request: Request, exc: CustomException):
    return HTTPException(
        status_code=404,
        detail={"message": f"some problem"},
    )


@app.get("/{name}")
async def read_unicorns(name: str):
    if name_true:
        logger.info("some info")
    raise UnicornException(name=name, info="some info")

1. 定义带日志级别的自定义异常类 #

直接继承FastAPI的HTTPException，新增log_level参数用于指定日志级别，保留原类的所有特性：

from fastapi import HTTPException
import logging

class LoggedHTTPException(HTTPException):
    def __init__(
        self,
        status_code: int,
        detail: str | dict = None,
        headers: dict[str, str] | None = None,
        log_level: int = logging.ERROR  # 默认日志级别为ERROR
    ):
        super().__init__(status_code=status_code, detail=detail, headers=headers)
        self.log_level = log_level

2. 编写全局异常处理器 #

对接现有日志系统，在处理器中根据异常指定的级别输出日志，同时返回标准HTTP响应：

from fastapi import Request, JSONResponse
from log_file import Logger  # 导入自定义日志实例

logger = Logger.logger

@app.exception_handler(LoggedHTTPException)
async def logged_exception_handler(request: Request, exc: LoggedHTTPException):
    # 按指定级别记录日志，包含请求上下文便于排查
    logger.log(
        exc.log_level,
        f"请求异常: {request.method} {request.url.path} - 状态码: {exc.status_code}, 详情: {exc.detail}"
    )
    # 返回标准HTTP响应，保留原detail字段
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.detail},
        headers=exc.headers
    )

3. 在接口中使用自定义异常 #

在需要抛出异常的场景，直接使用LoggedHTTPException，可灵活指定日志级别：

@app.get("/vehicle/all")
async def get_all_vehicles():
    # 示例：抛出404异常，指定日志级别为ERROR
    raise LoggedHTTPException(
        status_code=404,
        detail="车辆列表不存在",
        log_level=logging.ERROR
    )

@app.get("/user/{user_id}")
async def get_user(user_id: int):
    # 示例：抛出400异常，指定日志级别为WARNING
    if user_id <= 0:
        raise LoggedHTTPException(
            status_code=400,
            detail={"user_id": "必须大于0"},
            log_level=logging.WARNING
        )
    # 正常业务逻辑...

关键说明 #

• 继承HTTPException确保响应格式完全符合FastAPI标准，无需重新构建响应结构
• log_level参数支持logging模块所有级别（如logging.INFO、logging.CRITICAL等）
• 日志内容包含请求方法、路径等上下文信息，便于定位问题
• 完美对接现有自定义日志系统，无需修改原有日志配置

内容的提问来源于stack exchange，提问作者MichalK