FastAPI:(6)错误处理


FastAPI:(6)错误处理

概述

文章概念关系

graph TD
    A[错误处理] --> B[HTTPException]
    A --> C[自定义异常类]
    A --> D[异常处理器]
    D --> E[注册处理器]
    E --> F[@app.exception_handler]
    B --> G[状态码]
    B --> H[detail]
    B --> I[headers]
    C --> J[UnicornException]
    D --> K[RequestValidationError]
    K --> L[请求体]
    K --> M[覆盖默认处理器]
    D --> N[复用默认处理器]
    N --> O[fastapi.exception_handlers]

错误处理

错误处理是指在应用运行过程中对可能发生的异常或错误进行捕获、统一响应和返回格式化错误信息的机制。FastAPI 支持对内建异常(如 HTTPException)的处理,以及自定义异常类型的注册和响应方式。良好的错误处理机制可以提升 API 的鲁棒性、用户体验和调试效率。

重要特征

  • 特征1基于 HTTPException 或自定义异常类
    FastAPI 默认支持 fastapi.HTTPException,也允许定义自己的异常类和异常处理器。
  • 特征2响应内容结构化,可自定义状态码、消息和返回体
    错误响应可包括 status_codedetailheaders 等字段。
  • 特征3通过 @app.exception_handler 注册处理器
    可针对特定异常类型定义处理逻辑,统一捕获并生成响应。
  • 特征4与路径操作函数逻辑解耦
    使主业务代码更简洁清晰,错误处理集中统一。

订单查询异常处理(正例)

现象:
在一个电商平台中,用户根据订单 ID 查询订单详情。若订单不存在,则抛出 HTTPException(404),返回“订单未找到”的错误响应。

特征对比

特征 是否满足
使用 HTTPException 或自定义异常 ✅ 使用内建 HTTPException
响应结构规范(状态码、消息) ✅ 设置 404 + detail
捕获逻辑清晰明确 ✅ 异常直接由业务逻辑触发
与主业务逻辑解耦 ✅ 查询失败即抛异常,无需后续处理判断
from fastapi import FastAPI, HTTPException

app = FastAPI()

fake_orders = {"123": "订单123内容", "456": "订单456内容"}

@app.get("/order/{order_id}")
async def get_order(order_id: str):
    if order_id not in fake_orders:
        raise HTTPException(status_code=404, detail="订单未找到")
    return {"order_id": order_id, "content": fake_orders[order_id]}

HTTPException 是额外包含了和 API 有关数据的常规 Python 异常。因为是 Python 异常,所以不能 return,只能 raise

如在调用「路径操作函数」里的工具函数时,触发了 HTTPException,FastAPI 就不再继续执行_路径操作函数_中的后续代码,而是立即终止请求,并把 HTTPException 的 HTTP 错误发送至客户端。

触发 HTTPException 时,可以用参数 detail 传递任何能转换为 JSON 的值,不仅限于 str。还支持传递 dictlist 等数据结构。FastAPI 能自动处理这些数据,并将之转换为 JSON。

return

现象:
当用户输入非法请求时,系统返回如下内容:

return {"error": "invalid input"}

并未使用 HTTP 状态码或异常处理机制。

特征对比

特征 是否满足
使用 HTTPException 或自定义异常 ❌ 直接 return,未抛异常
响应结构标准 ❌ 未设置状态码,结构不规范
捕获逻辑统一 ❌ 所有错误需开发者手动处理
与主业务解耦 ❌ 错误处理混杂在主逻辑中,易出错

自定义响应头

有些场景下要为 HTTP 错误添加自定义响应头。例如,出于某些方面的安全需要。一般情况下可能不会需要在代码中直接使用响应头。但对于某些高级应用场景,还是需要添加自定义响应头:

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "There goes my error"}, # 添加响应头
        )
    return {"item": items[item_id]}

自定义异常处理器

添加自定义处理器,要使用 Starlette 的异常工具。假设要触发的自定义异常叫作 UnicornException。且需要 FastAPI 实现全局处理该异常。此时,可以用 @app.exception_handler() 添加自定义异常控制器:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

请求 /unicorns/yolo 时,路径操作会触发 UnicornException
但该异常将会被 unicorn_exception_handler 处理。
接收到的错误信息清晰明了,HTTP 状态码为 418,JSON 内容如下:

{"message": "Oops! yolo did something. There goes a rainbow..."}

覆盖请求验证异常

「请求中包含无效数据」时,FastAPI 内部会触发 RequestValidationError。该异常也内置了默认异常处理器。

覆盖默认异常处理器时需要导入 RequestValidationError,并用 @app.excption_handler(RequestValidationError) 装饰异常处理器。这样,异常处理器就可以接收 Request 与异常。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)

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

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

#e 覆盖 HTTPException 处理器

只为错误返回纯文本响应,而不是返回 JSON 格式的内容。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import PlainTextResponse
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request, exc):
    return PlainTextResponse(str(exc.detail), status_code=exc.status_code)


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


@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

RequestValidationError

RequestValidationError 包含其接收到的无效数据请求的 body 。
开发时,可以用这个请求体生成日志、调试错误,并返回给用户。

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

app = FastAPI()


@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(), "body": exc.body}),
    )

class Item(BaseModel):
    title: str
    size: int

@app.post("/items/")
async def create_item(item: Item):
    return item

发送无效item

{
  "title": "towel",
  "size": "XL"
}

收到422,Error: Unprocessable Entity的body响应体。

```json
{
  "detail": [
    {
      "type": "int_parsing",
      "loc": [
        "body",
        "size"
      ],
      "msg": "Input should be a valid integer, unable to parse string as an integer",
      "input": "XL"
    }
  ],
  "body": {
    "title": "towel",
    "size": "XL"
  }
}

复用FastAPI异常处理器

FastAPI 支持先对异常进行某些处理,然后再使用 FastAPI 中处理该异常的默认异常处理器。从 fastapi.exception_handlers 中导入要复用的默认异常处理器,可以在处理异常之后再复用默认的异常处理器。

from fastapi import FastAPI, HTTPException
from fastapi.exception_handlers import (
    http_exception_handler,
    request_validation_exception_handler,
)
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException

app = FastAPI()

@app.exception_handler(StarletteHTTPException)
async def custom_http_exception_handler(request, exc):
    print(f"OMG! An HTTP error!: {repr(exc)}")
    return await http_exception_handler(request, exc)

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    print(f"OMG! The client sent invalid data!: {exc}")
    return await request_validation_exception_handler(request, exc)

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 3:
        raise HTTPException(status_code=418, detail="Nope! I don't like 3.")
    return {"item_id": item_id}

文章作者: Hkini
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 Hkini !
评论
  目录