FastAPI 错误处理
在 API 开发中,需要向客户端返回错误信息。FastAPI 提供了 HTTPException 来抛出 HTTP 错误,也支持自定义异常处理器来处理更复杂的错误场景。
使用 HTTPException
当需要返回错误响应时,抛出 HTTPException:
实例
from fastapi import FastAPI, HTTPException
app = FastAPI()
items = {"foo": "The Foo Wrestlers", "bar": "The Bar Fighters"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
# 资源不存在时,抛出 404 错误
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
app = FastAPI()
items = {"foo": "The Foo Wrestlers", "bar": "The Bar Fighters"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
# 资源不存在时,抛出 404 错误
raise HTTPException(status_code=404, detail="Item not found")
return {"item": items[item_id]}
访问 http://127.0.0.1:8000/items/42 时,返回:
{
"detail": "Item not found"
}
HTTP 状态码为 404。
注意这里使用的是
raise而不是return。HTTPException是一个异常,抛出后 FastAPI 会捕获它并返回对应的 HTTP 错误响应。这与 Python 的异常处理机制一致。
HTTPException 参数
| 参数 | 类型 | 说明 |
|---|---|---|
status_code | int | HTTP 状态码(必填) |
detail | Any | 错误详情,可以是字符串、字典、列表等 |
headers | dict | None | 额外的响应头(可选) |
自定义错误详情
detail 参数可以是任意类型,不局限于字符串:
实例
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id == "invalid":
# detail 可以是字典,包含更多错误信息
raise HTTPException(
status_code=400,
detail={
"error_code": "INVALID_ID",
"message": "商品ID格式不正确",
"hint": "请使用字母数字组合的ID"
}
)
return {"item_id": item_id}
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id == "invalid":
# detail 可以是字典,包含更多错误信息
raise HTTPException(
status_code=400,
detail={
"error_code": "INVALID_ID",
"message": "商品ID格式不正确",
"hint": "请使用字母数字组合的ID"
}
)
return {"item_id": item_id}
添加自定义响应头
某些场景下需要在错误响应中添加自定义响应头:
实例
from fastapi import FastAPI, HTTPException
app = FastAPI()
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
if item_id == "invalid":
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "There goes my error"}, # 自定义响应头
)
return {"item": item_id}
app = FastAPI()
@app.get("/items-header/{item_id}")
async def read_item_header(item_id: str):
if item_id == "invalid":
raise HTTPException(
status_code=404,
detail="Item not found",
headers={"X-Error": "There goes my error"}, # 自定义响应头
)
return {"item": item_id}
在需要限制客户端请求频率(限流)的场景中,可以在错误响应头中添加
Retry-After头,告诉客户端何时可以重试。
自定义异常处理器
你可以为自定义异常或内置异常注册处理器,统一处理错误响应格式:
实例
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 wrong"},
)
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
if name == "yolo":
# 抛出自定义异常
raise UnicornException(name=name)
return {"unicorn_name": name}
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 wrong"},
)
@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
if name == "yolo":
# 抛出自定义异常
raise UnicornException(name=name)
return {"unicorn_name": name}
覆盖默认异常处理器
FastAPI 有默认的异常处理器,你可以覆盖它们来自定义错误响应格式:
实例
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
# 覆盖 HTTP 异常处理器
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": f"HTTP error: {exc.detail}"},
)
# 覆盖请求校验异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={"error": "数据校验失败", "details": exc.errors()},
)
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
app = FastAPI()
# 覆盖 HTTP 异常处理器
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"error": f"HTTP error: {exc.detail}"},
)
# 覆盖请求校验异常处理器
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
return JSONResponse(
status_code=422,
content={"error": "数据校验失败", "details": exc.errors()},
)
RequestValidationError是 FastAPI 在请求数据校验失败时抛出的异常。覆盖它的处理器可以自定义校验错误的响应格式。
重定向
使用 RedirectResponse 实现重定向:
实例
from fastapi import FastAPI
from fastapi.responses import RedirectResponse
app = FastAPI()
@app.get("/items/")
async def read_items():
return {"items": ["foo", "bar"]}
@app.get("/redirect")
async def redirect():
# 重定向到 /items/
return RedirectResponse(url="/items/")
from fastapi.responses import RedirectResponse
app = FastAPI()
@app.get("/items/")
async def read_items():
return {"items": ["foo", "bar"]}
@app.get("/redirect")
async def redirect():
# 重定向到 /items/
return RedirectResponse(url="/items/")
自定义响应头和状态码
使用 JSONResponse 自定义响应头和状态码:
实例
from fastapi import FastAPI
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
content = {"item_id": item_id}
headers = {"X-Custom-Header": "custom-header-value"}
return JSONResponse(content=content, headers=headers)
from fastapi.responses import JSONResponse
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
content = {"item_id": item_id}
headers = {"X-Custom-Header": "custom-header-value"}
return JSONResponse(content=content, headers=headers)
小结
- 使用
raise HTTPException返回 HTTP 错误响应 detail参数可以是任意类型(字符串、字典、列表等)- 使用
@app.exception_handler注册自定义异常处理器 - 可以覆盖 FastAPI 默认的异常处理器来统一错误格式
- 使用
RedirectResponse实现重定向
