异常处理（Exception Handling）：全局异常捕获与自定义错误响应

📂 所属阶段：第二阶段 — 进阶黑科技（核心篇）
🔗 相关章节：中间件应用 · APIRouter 模块化

1. FastAPI 异常处理机制

1.1 FastAPI 的异常体系

HTTPException（业务异常）
    ├── 400 Bad Request
    ├── 401 Unauthorized
    ├── 403 Forbidden
    ├── 404 Not Found
    ├── 422 Validation Error
    └── 自定义 5xx

RequestValidationError（请求验证异常）
StarletteHTTPException（基类）
Python 内置异常（ValueError, KeyError, etc.）

1.2 抛出 HTTPException

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"1": "Apple", "2": "Banana", "3": "Orange"}

@app.get("/items/{item_id}")
async def get_item(item_id: str):
    if item_id not in items:
        # 抛出 HTTPException，FastAPI 自动转为 JSON 响应
        raise HTTPException(
            status_code=404,
            detail=f"Item {item_id} not found",  # 自定义错误信息
            headers={"X-Error": "ItemMissing"}
        )
    return {"item_id": item_id, "name": items[item_id]}

FastAPI 会自动将其转为：

{
  "detail": "Item 99 not found"
}

2. 全局异常处理器

2.1 为什么需要全局处理？

  • 路由中未捕获的异常会导致 500 Internal Server Error
  • 不同路由的异常格式不统一
  • 线上暴露异常栈信息有安全风险

2.2 注册全局异常处理器

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import traceback
import logging

logger = logging.getLogger(__name__)
app = FastAPI()

# ── 1. HTTPException 处理器 ────────────────────────
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code,
            "message": exc.detail,
            "path": str(request.url),
        }
    )

# ── 2. 请求验证异常处理器（参数校验失败）─────────────
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"])
        errors.append({"field": field, "message": error["msg"]})

    return JSONResponse(
        status_code=422,
        content={
            "code": 422,
            "message": "参数校验失败",
            "errors": errors,
        }
    )

# ── 3. 通用异常处理器（兜底）───────────────────────
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 生产环境：记录日志，返回通用错误
    logger.error(f"Unhandled error: {exc}\n{traceback.format_exc()}")

    return JSONResponse(
        status_code=500,
        content={
            "code": 500,
            "message": "服务器内部错误，请稍后重试",
            # 开发环境可以返回更多信息：
            # "detail": str(exc),
            # "traceback": traceback.format_exc(),
        }
    )

2.3 自定义业务异常类

# exceptions.py — 统一的自定义异常
class AppException(Exception):
    """基础业务异常"""
    def __init__(self, message: str, code: int = 400, details: dict = None):
        self.message = message
        self.code = code
        self.details = details or {}
        super().__init__(message)

class NotFoundException(AppException):
    """资源不存在"""
    def __init__(self, resource: str, identifier: str):
        super().__init__(
            message=f"{resource} '{identifier}' 不存在",
            code=404,
            details={"resource": resource, "identifier": identifier}
        )

class UnauthorizedException(AppException):
    """未授权"""
    def __init__(self, reason: str = "请先登录"):
        super().__init__(message=reason, code=401)

class ForbiddenException(AppException):
    """权限不足"""
    def __init__(self):
        super().__init__(message="权限不足，无法访问该资源", code=403)

class RateLimitException(AppException):
    """请求过于频繁"""
    def __init__(self, retry_after: int = 60):
        super().__init__(
            message="请求过于频繁，请稍后重试",
            code=429,
            details={"retry_after": retry_after}
        )
# 注册业务异常处理器
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    content = {
        "code": exc.code,
        "message": exc.message,
    }
    if exc.details:
        content["details"] = exc.details

    response = JSONResponse(status_code=exc.code, content=content)

    if isinstance(exc, RateLimitException):
        response.headers["Retry-After"] = str(exc.details.get("retry_after", 60))

    return response

2.4 使用自定义异常

from exceptions import NotFoundException, UnauthorizedException

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await db.find_user(user_id)
    if not user:
        raise NotFoundException("用户", str(user_id))
    return user

@app.delete("/articles/{article_id}")
async def delete_article(article_id: int, token: str = Header(...)):
    user = await auth.verify(token)
    article = await db.find_article(article_id)

    if article.author_id != user.id:
        raise UnauthorizedException("无权删除他人文章")

    await db.delete(article)
    return {"deleted": article_id}

3. 统一响应格式

3.1 响应模型封装

from pydantic import BaseModel
from typing import TypeVar, Generic, Optional, Any

T = TypeVar("T")

class ApiResponse(BaseModel, Generic[T]):
    """统一响应格式"""
    code: int = 200
    message: str = "success"
    data: Optional[T] = None
    error: Optional[str] = None

    class Config:
        json_schema_extra = {
            "example": {
                "code": 200,
                "message": "success",
                "data": {"id": 1, "name": "Alice"}
            }
        }

def success_response(data: Any = None, message: str = "success"):
    return ApiResponse(code=200, message=message, data=data)

def error_response(code: int, message: str, error: str = None):
    return ApiResponse(code=code, message=message, error=error)

3.2 在路由中使用

from fastapi import APIRouter
from typing import List

router = APIRouter()

@router.get("/users", response_model=ApiResponse[List[dict]])
async def list_users():
    users = await db.query_users()
    return success_response(data=users, message="获取用户列表成功")

@router.get("/users/{user_id}", response_model=ApiResponse[dict])
async def get_user(user_id: int):
    user = await db.find_user(user_id)
    if not user:
        return error_response(404, "用户不存在")
    return success_response(data=user)

4. 异常处理与日志

4.1 结构化日志

import structlog
import logging

# 配置结构化日志
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.stdlib.BoundLogger,
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

@app.exception_handler(Exception)
async def log_exception(request: Request, exc: Exception):
    logger.error(
        "request_failed",
        method=request.method,
        path=str(request.url.path),
        error_type=type(exc).__name__,
        error_message=str(exc),
        traceback=traceback.format_exc(),
    )

    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "Internal server error"}
    )

4.2 按异常类型差异化处理

@app.exception_handler(Exception)
async def smart_exception_handler(request: Request, exc: Exception):
    if isinstance(exc, AppException):
        # 业务异常：直接返回业务定义的错误
        return JSONResponse(
            status_code=exc.code,
            content={"code": exc.code, "message": exc.message}
        )

    if isinstance(exc, ValueError):
        # 业务参数错误 → 400
        return JSONResponse(400, {"code": 400, "message": str(exc)})

    if isinstance(exc, ConnectionError):
        # 服务不可用 → 503
        logger.error("service_unavailable", error=str(exc))
        return JSONResponse(503, {"code": 503, "message": "服务暂时不可用"})

    # 未知异常 → 500 + 记录
    logger.exception("unhandled_error")
    return JSONResponse(500, {"code": 500, "message": "服务器错误"})

5. 404 处理

5.1 全局 404 处理器

from starlette.requests import Request
from starlette.responses import JSONResponse

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    if exc.status_code == 404:
        return JSONResponse(
            status_code=404,
            content={
                "code": 404,
                "message": f"API 路径 '{request.url.path}' 不存在",
                "suggestion": "请检查 URL 是否正确，或访问 /docs 查看可用接口"
            }
        )
    # 其他 HTTP 异常走通用处理
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": exc.detail}
    )

6. 验证错误详解

6.1 RequestValidationError 结构

# 触发 422 验证错误
@app.get("/calculate")
async def calculate(x: int, y: int):
    return {"result": x / y}

# GET /calculate?x=10&y=0 → 除零错误（Python 异常，非验证异常）
# GET /calculate?x=abc&y=2 → 参数类型验证错误（RequestValidationError）

验证错误响应示例：

{
  "code": 422,
  "message": "参数校验失败",
  "errors": [
    {
      "field": "body.price",
      "message": "Input should be greater than or equal to 0"
    }
  ]
}

6.2 自定义验证错误消息

from pydantic import BaseModel, field_validator

class CreateUser(BaseModel):
    name: str
    email: str
    age: int

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("用户名不能为空")
        return v.strip()

    @field_validator("email")
    @classmethod
    def email_format(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("邮箱格式不正确")
        return v.lower()

    @field_validator("age")
    @classmethod
    def age_range(cls, v: int) -> int:
        if v < 0 or v > 150:
            raise ValueError("年龄必须在 0-150 之间")
        return v

@app.post("/users")
async def create_user(user: CreateUser):
    return {"created": user.model_dump()}

7. 异常处理最佳实践

✅ 推荐：
1. 自定义业务异常类，统一错误码和消息格式
2. 异常处理器返回统一的 JSON 结构
3. 生产环境不要暴露异常堆栈
4. 记录详细日志供排查
5. HTTPException 用于"预期内的错误"
6. 全局兜底异常处理器防止意外 500 泄漏

❌ 避免：
1. 路由中用 try/except 吞掉所有异常但不处理
2. 在 except 中 print 而不记录日志
3. 混用不同的错误响应格式
4. 过于宽泛的 except Exception

8. 小结

异常类型触发场景FastAPI 默认响应
HTTPException业务逻辑（资源不存在、权限不足）{"detail": "..."}
RequestValidationError参数校验失败详细错误位置列表
Exception所有未捕获异常500（需自定义兜底）
自定义 AppException业务特定错误自定义结构

💡 核心原则：异常处理的目标是让客户端收到可预期的错误响应，同时让服务端记录足够排查问题的日志。

🔗 扩展阅读