一文解析Python FastAPI进行异常处理的详细方法

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in item_db:
        # 关键在这里：抛出带状态码和详情的异常
        raise HTTPException(
            status_code=404,
            detail="Item not found",
            headers={"X-Error": "ItemID-Missing"}
        )
    return {"item": item_db[item_id]}

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

app = FastAPI()

# 1. 先定义一个标准的错误响应模型
class ErrorResponse(BaseModel):
    code: int
    message: str
    detail: Optional[str] = None
    request_id: Optional[str] = None # 用于链路追踪

# 2. 捕获所有未处理异常的“总闸”
@app.exception_handler(Exception)
async def universal_exception_handler(request: Request, exc: Exception):
    # 获取请求ID，便于追踪（假设从中间件或header传入）
    request_id = request.headers.get("X-Request-ID", "unknown")
    
    # 这里可以根据exc的类型进行更精细的分类
    error_code = 500 # 默认内部错误
    message = "Internal Server Error"
    
    if isinstance(exc, ValueError):
        error_code = 400
        message = "Invalid input value"
    # ... 可以添加更多类型判断
    
    # 在生产环境，detail可能不返回具体堆栈，开发环境可以返回
    import os
    detail = traceback.format_exc() if os.getenv("ENV") == "development" else None
    
    return JSONResponse(
        status_code=error_code,
        content=ErrorResponse(
            code=error_code,
            message=message,
            detail=detail,
            request_id=request_id
        ).dict()
    )

# 3. 专门处理HTTPException，覆盖FastAPI默认行为
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            code=exc.status_code,
            message=exc.detail,
            request_id=request.headers.get("X-Request-ID", "unknown")
        ).dict(),
        headers=exc.headers
    )

# 定义自己的业务异常类
class BusinessError(Exception):
    def __init__(self, code: int, message: str, extra_data: dict = None):
        self.code = code # 业务错误码，如 1001
        self.message = message
        self.extra_data = extra_data or {}

# 定义几个具体的业务异常
class InsufficientBalanceError(BusinessError):
    def __init__(self, current_balance: float, required_amount: float):
        super().__init__(
            code=1001,
            message="Insufficient balance",
            extra_data={
                "current_balance": current_balance,
                "required_amount": required_amount
            }
        )

class ActivityExpiredError(BusinessError):
    def __init__(self, activity_id: str, expire_time: str):
        super().__init__(
            code=1002,
            message="Activity has expired",
            extra_data={"activity_id": activity_id, "expire_time": expire_time}
        )

# 为自定义业务异常注册处理器
@app.exception_handler(BusinessError)
async def business_exception_handler(request: Request, exc: BusinessError):
    return JSONResponse(
        status_code=422, # 或用200，但body里表明错误，看前端约定
        content={
            "success": False,
            "error": {
                "code": exc.code,
                "message": exc.message,
                **exc.extra_data # 展开额外数据，前端可以直接用
            }
        }
    )

# 在路由中使用
@app.post("/purchase")
async def make_purchase(user_id: int, amount: float):
    user_balance = get_balance(user_id)
    if user_balance < amount:
        # 抛出业务异常，而不是简单的HTTP 400
        raise InsufficientBalanceError(
            current_balance=user_balance,
            required_amount=amount
        )
    # ... 购买逻辑

from fastapi import WebSocket, WebSocketException

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            # 一些业务验证
            if data.get("type") not in VALID_TYPES:
                # 抛出WebSocketException，指定关闭码和原因
                raise WebSocketException(
                    code=1008, # 1008表示政策违规
                    reason="Invalid message type received"
                )
            # ... 处理消息
    except WebSocketException as e:
        # 这里其实raise之后，FastAPI会帮你关闭连接
        raise
    except Exception as e:
        # 其他未知异常，也以WebSocketException形式关闭
        raise WebSocketException(code=1011, reason=f"Internal error: {str(e)}")