Python框架篇(4):FastApi-错误处理

@提示: 微信搜索【猿码记】回复 【fastapi】即可获取源码信息~

1.验证错误

1.1 默认返回

当传参不符合模型验证规则时，默认错误信息和格式返回如下:

{
  "detail": [
    {
      "loc": [
        "body",
        "age"
      ],
      "msg": "ensure this value is greater than 18",
      "type": "value_error.number.not_gt",
      "ctx": {
        "limit_value": 18
      }
    },
    {
      "loc": [
        "body",
        "likes"
      ],
      "msg": "ensure this value has at least 2 items",
      "type": "value_error.list.min_items",
      "ctx": {
        "limit_value": 2
      }
    }
  ]
}

在上篇文章 Python框架篇(3):FastApi-响应模型中，我们强调的是接口对外输出结构应该一致，所以即便参数错误应该输出以下结构:

{
    "code":-1,
    "msg":"具体错误信息...",
    "data": null,
    "additional":{
        "time":"2023-12-04 19:00:23",
        "trace_id":"cc1b12a5dfee26a7dcc29fe47dcfbde0"
    }
}

按照官方文档说法，我们需要自定义错误处理器，并以此来覆盖框架默认的异常处理器，参数验证错误处理器默认走的是RequestValidationError,所以覆盖它就行，下面是实现步骤

1.2 自定义处理器

新建包app/errors，并新增文件validation_error.py，文件内容如下:

from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from app.types import response
from fastapi.encoders import jsonable_encoder


async def validationExceptionHandler(request: Request, exc: RequestValidationError):
    """ 自定义参数验证异常错误"""
    errMsg = ""
    for error in exc.errors():
        errMsg += ".".join(error.get("loc")) + ":" + error.get("msg") + ";"
    # 这里response.ResponseFail是上篇文章中的内容
    return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder(response.ResponseFail(errMsg)))

在包app/errors/__init__.py引用,并封装统一注册方法：

from fastapi import FastAPI
from .validation_error import validationExceptionHandler
from fastapi.exceptions import RequestValidationError


def registerCustomErrorHandle(server: FastAPI):
    """ 统一注册自定义错误处理器"""
    # 注册参数验证错误,并覆盖模式RequestValidationError
    server.add_exception_handler(RequestValidationError, validationExceptionHandler)

1.3 注册&覆盖

在main.py中调用registerCustomErrorHandle

# 引入
from app import errors

server = FastAPI(redoc_url=None, docs_url="/apidoc", title="FastAPI学习")
# 注册自定义错误处理器
errors.registerCustomErrorHandle(server)
...

@注:这里把之前的变量app改成server

1.4 验证

2.路由错误

路由错误常见的一般分为以下两种:

• 404：访问不存在的接口地址；
• 405： 接口定义的请求方式是POST,当时使用GET 方式请求时；

2.1 默认返回

// 当访问不存在路由时
{
    "detail": "Not Found"
}
// 当访问方式不对时
{
    "detail": "Method Not Allowed"
}

2.2 自定义处理器

新建包app/errors，并新增文件http_error.py，文件内容如下:

from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from starlette.exceptions import HTTPException
from app.types import response

async def httpExceptionHandler(request, exc: HTTPException) -> JSONResponse:
    """自定义处理HTTPException"""
    print("request:", request)
    print("status_code:", exc.status_code)
    if exc.status_code == status.HTTP_404_NOT_FOUND:
        # 处理404错误
        return JSONResponse(
            content=jsonable_encoder(response.ResponseFail("接口路由不存在~")),
            status_code=status.HTTP_200_OK,
        )
    elif exc.status_code == status.HTTP_405_METHOD_NOT_ALLOWED:
        # 处理405错误
        return JSONResponse(
            content=jsonable_encoder(response.ResponseFail("请求方式错误，请查看文档确认~")),
            status_code=status.HTTP_200_OK,
        )
    else:
        return JSONResponse(
            content=jsonable_encoder(response.ResponseFail(str(exc))),
            status_code=status.HTTP_200_OK,
        )

2.3 注册&覆盖

修改app/errors/__init__.py文件中的统一注册方法registerCustomErrorHandle：

from fastapi import FastAPI
from .validation_error import validationExceptionHandler
from .http_error import httpExceptionHandler
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

def registerCustomErrorHandle(server: FastAPI):
    """ 统一注册自定义错误处理器 """
    # 注册参数验证错误,并覆盖模式RequestValidationError
    server.add_exception_handler(RequestValidationError, validationExceptionHandler)
    # 错误处理StarletteHTTPException
    server.add_exception_handler(StarletteHTTPException, httpExceptionHandler)

@注意:这里覆盖的错误是:starlette.exceptions包中的HTTPException,不是这个包fastapi.exceptions,否则不会生效！

2.4 验证

// 当访问不存在路由时
{
    "code": -1,
    "msg": "接口路由不存在~",
    "data": null,
    "additional": {
        "time": "2023-12-08 19:45:28",
        "trace_id": "0c9990bb9292dc39909c6ebefc0a8684"
    }
}


// 当访问方式不对时
{
    "code": -1,
    "msg": "请求方式错误，请查看文档确认~",
    "data": null,
    "additional": {
        "time": "2023-12-08 19:45:55",
        "trace_id": "816d92949542040bfdb63922fcdb6ae9"
    }
}

3.翻译参数错误

在上面验证错误使用中，我们可以看到返回的错误信息是英文的，那么怎么翻译成中文呢，在官方文档中未找到使用方法，后来在网上搜到需要使用error_msg_templates这个属性，按照网上示例发现未为生效，后来在pydantic官网又看到这个属性居然在后来的版本中删除了…. :joy:

官方文档: Changes to config:https://docs.pydantic.dev/latest/migration/#changes-to-config

找不到正规途径，只能使用野蛮方法了….

3.1 分析错误对象

从这个错误对象，我们可以简单看出关键节点：type(错误类型)、loc(字段)、ctx(限制信息)，

3.2 定义错误模版

在包app/config，并新增文件validate_template_config.py，文件内容如下:

# 错误模版
validateChineseDict = {
    "value_error.number.not_gt": "{},值不能大于:{}",
    "value_error.number.not_ge": "{},值不能小于等于:{}",
    "value_error.list.min_items": "{},元素个数至少为:{}",
    "value_error.str.regex": "{},不满足规则:{}"
}

# 关键词显示的错误
keyErrorChineseDict = {
    "phone": "手机号格式不正确~"
}

在app/config/__init__.py，引入变量配置，文件内容如下:

from .validate_template_config import validateChineseDict, keyErrorChineseDict

3.3 改造自定义处理器

修改文件app/errors/validation_error.py，修改后内容如下:

from fastapi import Request, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from app.types import response
from app import config
from fastapi.encoders import jsonable_encoder


async def validationExceptionHandler(request: Request, exc: RequestValidationError):
    """ 自定义参数验证异常错误"""
    errMsg = ""
    for error in exc.errors():
        fieldName = ".".join(error.get("loc"))
        errType = error.get("type")
        if errType in config.validateChineseDict:
            # 在定义错误模版中，并翻译出内容
            translateMsg = translate(fieldName, errType, error.get("ctx")) + "; "
            if translateMsg:
                errMsg += translateMsg
        else:
            # 不在定义模型，显示原始错误
            errMsg += ".".join(error.get("loc")) + "[" + error.get("type") + "]:" + error.get("msg") + "; "

    # 替换body.
    errMsg = errMsg.replace("body.", "")
    # 返回
    return JSONResponse(status_code=status.HTTP_200_OK, content=jsonable_encoder(response.ResponseFail(errMsg)))


def translate(fieldName: str, errType: str, limitDict: dict) -> str:
    """ 翻译错误信息"""
    # 先判断是否满足关键词错误
    for k, v in config.keyErrorChineseDict.items():
        if fieldName.find(k) != -1:
            return v

    limitValList = limitDict.values()
    try:
        return config.validateChineseDict.get(errType).format(fieldName, *limitValList)
    except Exception as e:
        return ""

在python3.10/site-packages/pydantic/errors.py文件中，定义了经常使用的错误模版，可以跟结合本地实际情况，查看具体代码：

3.4 验证结果

@注意:上面方法，只是提供一种解决思路，如果有更好的方式，喜欢能留言通知，一起学习~

4.系统错误

在厉害的程序员都无法避免程序错误，特别是运行中遇到的异常。如果遇到异常时,我们不希望用户看到500，或者页面崩溃。

4.1 默认返回

4.2 自定义处理器

新建包app/errors，并新增文件app_error.py，文件内容如下:

from fastapi import status
from fastapi.encoders import jsonable_encoder
from fastapi.requests import Request
from fastapi.responses import JSONResponse
from app.types import response


async def appExceptionHandler(request: Request, exc: Exception):
    """自定义全局系统错误"""
    return JSONResponse(
        content=jsonable_encoder(response.ResponseFail("系统运行异常,稍后重试~")),
        status_code=status.HTTP_200_OK,
    )

4.3 注册&覆盖

修改app/errors/__init__.py文件中的统一注册方法registerCustomErrorHandle：

def registerCustomErrorHandle(server: FastAPI):
    """ 统一注册自定义错误处理器 """
    # 注册参数验证错误,并覆盖模式RequestValidationError
    server.add_exception_handler(RequestValidationError, validationExceptionHandler)
    # 错误处理StarletteHTTPException
    server.add_exception_handler(StarletteHTTPException, httpExceptionHandler)
    # 自定义全局系统错误
    server.add_exception_handler(Exception, appExceptionHandler)

4.4 验证