FastAPI错误处理:HTTP状态码与异常捕获实战

在API开发中,错误处理是确保系统健壮性和用户体验的关键。当用户请求出现问题(比如参数错误、资源不存在)时,良好的错误处理能让API返回清晰的提示,帮助调用方快速定位问题。FastAPI提供了简洁高效的错误处理机制,让我们既能灵活使用HTTP状态码,又能优雅地捕获和处理异常。

一、为什么需要错误处理?

想象一下,如果用户请求一个不存在的用户ID,而API直接崩溃或返回空白,用户会非常困惑。正确的错误处理可以:
- 向调用方返回明确的错误原因(比如“用户不存在”)
- 使用标准的HTTP状态码(如404、400),让前端或其他系统更容易解析
- 避免程序因未处理的异常而崩溃,保证服务稳定

二、HTTP状态码:API的“红绿灯”

HTTP状态码是错误处理的基础,FastAPI支持所有标准HTTP状态码。常见的状态码及使用场景如下:

状态码 含义 适用场景
200 成功 请求正常处理,返回数据
400 参数错误 请求参数不合法(如格式错误)
404 资源不存在 请求的资源(如用户、商品)不存在
401 未授权 需要登录但未提供有效凭证
403 禁止访问 权限不足
422 验证错误 请求参数格式正确但验证失败
500 服务器内部错误 服务端逻辑出错(如数据库连接失败)

FastAPI中使用状态码:在路由函数中,可通过returnraise直接指定状态码。例如:

from fastapi import FastAPI, HTTPException

app = FastAPI()

# 成功响应(默认200)
@app.get("/health")
def health_check():
    return {"status": "OK"}  # 自动返回200状态码

# 明确返回404状态码
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in [1, 2, 3]:  # 假设只有1、2、3存在
        return {"status": "error", "detail": "Item not found"}, 404  # 直接返回状态码

三、FastAPI异常捕获:主动“抛出”错误

FastAPI推荐使用HTTPException类主动抛出错误,它能直接指定状态码和错误信息,让错误处理更清晰。

1. 基础用法:HTTPException

from fastapi import HTTPException

@app.get("/users/{user_id}")
def get_user(user_id: int):
    # 模拟数据库查询
    users = {1: "Alice", 2: "Bob"}
    if user_id not in users:
        # 抛出HTTP异常,指定状态码和详细信息
        raise HTTPException(
            status_code=404,  # 资源不存在
            detail=f"用户ID {user_id} 不存在",  # 错误描述
            headers={"X-Error-Reason": "User not found"}  # 可选响应头
        )
    return {"user_id": user_id, "name": users[user_id]}

效果:当请求/users/999时,API会返回:

{
  "detail": "用户ID 999 不存在"
}

状态码为404,前端可根据状态码判断“资源不存在”。

2. 参数验证错误:自动返回422

FastAPI会自动处理参数格式错误(如类型不匹配),并返回422状态码。例如:

@app.get("/users/{user_id}")
def get_user(user_id: int):  # 强制user_id为整数
    # 如果用户传非整数(如字符串),FastAPI自动拦截并返回422
    return {"user_id": user_id}

效果:请求/users/abc时,API返回:

{
  "detail": [
    {
      "loc": ["path", "user_id"],
      "msg": "输入的值不是有效的整数",
      "type": "type_error.integer"
    }
  ]
}

状态码为422(验证错误),包含具体的错误字段和原因。

四、自定义异常:业务逻辑的“专属错误”

当业务逻辑出现错误(如“余额不足”“权限不足”)时,可自定义异常类,配合FastAPI的异常处理机制统一管理。

1. 定义自定义异常

class InsufficientFundsError(Exception):
    """自定义异常:余额不足"""
    def __init__(self, balance: float, needed: float):
        self.balance = balance  # 当前余额
        self.needed = needed    # 需要的金额
        self.detail = f"余额不足,当前余额: {balance},需要: {needed}"

2. 全局异常处理

通过@app.exception_handler注册自定义异常的处理函数,实现全局统一处理:

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

# 处理InsufficientFundsError异常
@app.exception_handler(InsufficientFundsError)
async def handle_insufficient_funds(request: Request, exc: InsufficientFundsError):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,  # 400状态码
        content={"detail": exc.detail}  # 返回自定义错误信息
    )

3. 在路由中使用自定义异常

@app.post("/withdraw")
def withdraw(amount: float):
    balance = 100.0  # 假设当前余额
    if amount > balance:
        # 抛出自定义异常
        raise InsufficientFundsError(balance=balance, needed=amount)
    return {"message": "转账成功", "balance": balance - amount}

效果:请求/withdraw?amount=200时,API返回:

{
  "detail": "余额不足,当前余额: 100.0,需要: 200.0"
}

五、全局错误处理:统一“兜底”策略

当出现未预料到的异常(如数据库连接失败)时,全局异常处理能避免重复代码,统一返回标准错误。

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

# 处理所有未捕获的异常(通用兜底)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 记录错误日志(生产环境建议使用logging)
    print(f"未处理异常: {str(exc)}")
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "服务器内部错误,请联系管理员"}
    )

六、最佳实践总结

  1. 用HTTPException处理HTTP标准错误:如404(资源不存在)、400(参数错误)、422(验证失败),直接抛出并指定状态码。
  2. 用自定义异常处理业务逻辑错误:如“余额不足”“权限不足”,配合exception_handler统一返回。
  3. 自动参数验证:FastAPI会自动处理类型错误,返回422状态码,无需手动捕获。
  4. 全局异常处理:通过@app.exception_handler统一兜底未捕获异常,避免重复代码。

示例代码整合

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

app = FastAPI()

# 模拟用户数据
users = {1: "Alice", 2: "Bob"}

# 自定义异常
class InsufficientFundsError(Exception):
    def __init__(self, balance: float, needed: float):
        self.balance = balance
        self.needed = needed
        self.detail = f"余额不足,当前余额: {balance},需要: {needed}"

# 全局异常处理:兜底所有未捕获异常
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    print(f"全局错误: {str(exc)}")
    return JSONResponse(
        status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        content={"detail": "服务器内部错误"}
    )

# 处理自定义异常:余额不足
@app.exception_handler(InsufficientFundsError)
async def handle_insufficient_funds(request: Request, exc: InsufficientFundsError):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={"detail": exc.detail}
    )

# 1. 获取用户(处理404)
@app.get("/users/{user_id}")
def get_user(user_id: int):
    if user_id not in users:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"用户ID {user_id} 不存在"
        )
    return {"user_id": user_id, "name": users[user_id]}

# 2. 模拟转账(处理自定义异常)
@app.post("/withdraw")
def withdraw(amount: float):
    balance = 100.0
    if amount > balance:
        raise InsufficientFundsError(balance=balance, needed=amount)
    return {"message": "转账成功", "remaining": balance - amount}

通过以上内容,你已经掌握了FastAPI错误处理的核心方法:用HTTP状态码规范错误类型,用HTTPException和自定义异常处理具体错误场景,并通过全局异常处理统一兜底。这些技巧能帮助你构建更健壮、友好的API服务。

Xiaoye