如何在 FastAPI 中自定义错误响应？

问题描述：

我有以下 FastAPI 后端：

from fastapi import FastAPI

app = FastAPI

class Demo(BaseModel):
    content: str = None
    
@app.post("/demo")
async def demoFunc(d:Demo):
    return d.content

问题是，当我向此 API 发送包含额外数据的请求时，例如：

data = {"content":"some text here"}aaaa

或者

data = {"content":"some text here"aaaaaa}

resp = requests.post(url, json=data)

在以下情况下，它会引发错误，状态代码为422 unprocessable entity错误，返回字段中存在实际（“此处有一些文本”）和额外（“aaaaa”）数据data = {"content":"some text here"}aaaa：

{
  "detail": [
    {
      "loc": [
        "body",
        47
      ],
      "msg": "Extra data: line 4 column 2 (char 47)",
      "type": "value_error.jsondecode",
      "ctx": {
        "msg": "Extra data",
        "doc": "{
  \"content\": \"some text here\"}aaaaa",
        "pos": 47,
        "lineno": 4,
        "colno": 2
      }
    }
  ]
}

我尝试将这一行放在app=FastAPI()try-catch 块中，但是它不起作用。有没有办法用自己的响应而不是上面提到的自动响应来处理这个问题？像这样：

{"error": {"message": "Invalid JSON body"},
                         "status": 0}

解决方案 1：

您传递的是无效的 JSON，因此服务器正确地响应了错误422 Unprocessable Entity。您的测试客户端根本无法运行，否则会抛出invalid syntax错误。因此，我猜您是通过 Swagger UI 提供的交互式自动文档在 上发布的请求/docs，并收到了相关的 422 错误。

如果您真正想要的是处理错误，为了自定义错误或其他内容，您可以覆盖请求验证异常处理程序，如文档中所述（请参阅此讨论，以及此答案和此答案，演示了如何RequestValidationError仅针对特定路线进行自定义）。

工作示例：

from fastapi import FastAPI, Body, Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import BaseModel

app = FastAPI()


class Demo(BaseModel):
    content: str = None


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content=jsonable_encoder({"detail": exc.errors(),  # optionally include the errors
                "body": exc.body,
                 "custom msg": {"Your error message"}}),
    )


@app.post("/demo")
async def some_func(d: Demo):
    return d.content

或者，您也可以返回PlainTextResponse自定义消息：

from fastapi.responses import PlainTextResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return PlainTextResponse(str(exc), status_code=422) 

解决方案 2：

我个人使用此代码将错误消息翻译成波斯语和西班牙语：

from logging import getLogger
from re import subn
from traceback import format_exc

from fastapi import (
    FastAPI,
    Request,
    status,
    Response,
)
from fastapi.exceptions import RequestValidationError

logger = getLogger(__name__)
validation_error_message_cache = {}

app = FastAPI()


@app.exception_handler(RequestValidationError)
async def handler_for_validation_error(
        request: Request,
        exc: RequestValidationError,
) -> Response:
    exe_error = exc.errors()
    try:
        for i in exe_error:
            i['msg'] = translate(
                error_msg=i['msg'],
                error_language='farsi',
            )
    except Exception:
        logger.warning(
            "Exception occurred when translating this error message: %s 
 %s",
            exe_error,
            format_exc(),
        )

    return Response(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "success": False,
            "data": None,
            "error": exe_error,
            "message": 'یه چیزی کز خورد!',
        }
    )


def translate(
        error_msg: str,
        error_language: str,
):
    cache_key = f"{error_language}_{error_msg}"
    if cache_key in validation_error_message_cache:
        return validation_error_message_cache[cache_key]

    for key, value in VALIDATION_REGEX_MESSAGE_DICT.items():
        try:
            output_msg, count = subn(
                key,
                value[error_language],
                error_msg,
            )
            if count:
                validation_error_message_cache[cache_key] = output_msg
                return output_msg

        except Exception:
            logger.warning(
                "Exception occurred when translating by this regex: %s 
 %s",
                key,
                format_exc(),
            )

    logger.warning("Cannot find translation for this error message: %s", error_msg)

    return error_msg


VALIDATION_REGEX_MESSAGE_DICT = {
    r'^String should have at least (?P<length>.+) characters$': {
        'farsi': r"متن باید حداقل g<length> حرف داشته باشد.",
        'spanish': r"La cadena debe tener al menos (?P<longitud>.+) caracteres.",
    },
    
    r'^String should have at most (?P<length>.+) characters$': {
        'farsi': r"متن میتواند حداکثر g<length> حرف داشته باشد.",
        'spanish': r"La cadena debe tener como máximo (?P<longitud>.+) caracteres.",
    },

    r'^String should match pattern (?P<pattern>.+)$': {
        'farsi': r"متن باید متناسب با این الگو باشد: g<pattern>",
        'spanish': r"La cadena debe coincidir con el patrón (?P<patrón>.+).",
    },
}

请考虑：

• VALIDATION_REGEX_MESSAGE_DICT可以改变并扩展以包含其他类型的错误消息。

• 这个解决方案几乎是由pydantic建议的。

• 其他键和值如ctx、input、loc和type也url可以通过此方法改变。

• Python 正则表达式足够快，但我使用缓存字典validation_error_message_cache来使这个处理程序更快。

• 有些错误可以在pydantic.v1.errors

解决方案 3：

当你发送带有主体的请求时，FastAPI 将尝试使用loads内置json模块中的函数反序列化主体，并且你收到的错误由此函数引发。

因此，如果您想自定义响应，则必须按照Chris所说的方式创建一个异常处理程序并进行处理。就像这样：

from fastapi import status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError


app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
   errors = exc.errors()
   response = []
   for error in errors:
      if error['type'] == 'value_error.jsondecode':
         response.append({'error': {'message': 'Invalid JSON body'}, 'status': 0})
      else:
      response.append(error)
   return JSONResponse(content=response, status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)