从零开始系统学习 Python 编程

Appearance

07-错误处理 ​

Python 3.11+

本章讲解 FastAPI 异常处理和自定义错误响应。

第一部分：HTTPException ​

1.1 实际场景 ​

请求的资源不存在时，需要返回 404 错误和友好的错误信息。

问题：如何返回 HTTP 错误响应？

1.2 基本使用 ​

python
from fastapi import FastAPI, HTTPException

app: FastAPI = FastAPI()

items_db: dict[int, str] = {1: "Item 1"}


@app.get("/items/{item_id}")
def read_item(item_id: int) -> str:
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]

1.3 自定义状态码 ​

python
from fastapi import status


@app.post("/users")
def create_user(username: str, email: str) -> dict[str, str]:
    if email in existing_emails:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Email already registered"
        )
    return {"id": "1", "email": email}

第二部分：自定义异常 ​

2.1 实际场景 ​

业务逻辑有特定的错误类型，如余额不足、库存不足等，需要统一的错误格式。

问题：如何定义业务异常？

2.2 定义异常类 ​

python
from fastapi import HTTPException


class ItemNotFoundException(HTTPException):
    def __init__(self, item_id: int):
        super().__init__(
            status_code=404,
            detail=f"Item {item_id} not found"
        )


class ValidationException(HTTPException):
    def __init__(self, field: str, message: str):
        super().__init__(
            status_code=422,
            detail={"field": field, "message": message}
        )

2.3 使用自定义异常 ​

python
@app.get("/items/{item_id}")
def read_item(item_id: int) -> str:
    if item_id not in items_db:
        raise ItemNotFoundException(item_id)
    return items_db[item_id]

第三部分：异常处理器 ​

3.1 实际场景 ​

所有错误响应需要统一的 JSON 格式，包含错误码、错误消息等。

问题：如何全局处理异常并格式化响应？

3.2 注册处理器 ​

python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app: FastAPI = FastAPI()


@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException) -> JSONResponse:
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": "Item Not Found", "message": exc.detail}
    )


@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        status_code=422,
        content={"error": "Validation Error", "details": exc.errors()}
    )

3.3 全局错误处理 ​

python
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    return JSONResponse(
        status_code=500,
        content={"error": "Internal Server Error", "message": str(exc)}
    )

第四部分：完整示例 ​

python
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel

app: FastAPI = FastAPI()


# 自定义异常
class BusinessException(Exception):
    def __init__(self, code: int, message: str):
        self.code: int = code
        self.message: str = message


# 异常处理器
@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException) -> JSONResponse:
    return JSONResponse(
        status_code=exc.code,
        content={"error": exc.message}
    )


@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(
        status_code=422,
        content={
            "error": "Validation Error",
            "details": exc.errors()
        }
    )


# 模型
class Item(BaseModel):
    name: str
    price: float


# 路由
items_db: dict[int, dict[str, str | float]] = {1: {"name": "Apple", "price": 1.5}}


@app.get("/items/{item_id}")
def get_item(item_id: int) -> dict[str, str | float]:
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="Item not found")
    return items_db[item_id]


@app.post("/items")
def create_item(item: Item) -> dict[str, int | dict]:
    if item.price < 0:
        raise BusinessException(400, "Price cannot be negative")
    item_id: int = len(items_db) + 1
    items_db[item_id] = item.model_dump()
    return {"id": item_id, "item": item.model_dump()}


@app.get("/error")
def trigger_error():
    raise BusinessException(418, "I'm a teapot")

第五部分：L3 专家层 ​

5.1 Starlette 的异常处理链 ​

FastAPI 的异常处理基于 Starlette 的 ExceptionHandlerMiddleware，采用从具体到通用的匹配策略：

异常抛出
    │
    ▼
┌──────────────────────────────────┐
│  ExceptionHandlerMiddleware      │
│  ┌────────────────────────────┐  │
│  │  1. 精确匹配               │  │
│  │     ItemNotFoundException   │  │────→ 注册了处理器？是 → 执行 → 返回响应
│  └────────────────────────────┘  │
│  ┌────────────────────────────┐  │
│  │  2. 继承链向上查找          │  │
│  │     HTTPException ──→ ValueError  │────→ 注册了处理器？是 → 执行 → 返回响应
│  └────────────────────────────┘  │
│  ┌────────────────────────────┐  │
│  │  3. 422 Validation Error   │  │────→ Pydantic 校验失败时触发
│  └────────────────────────────┘  │
│  ┌────────────────────────────┐  │
│  │  4. 全局 Exception 处理器   │  │────→ 最后的兜底
│  └────────────────────────────┘  │
│  ┌────────────────────────────┐  │
│  │  5. 默认行为               │  │────→ 500 Internal Server Error
│  └────────────────────────────┘  │
└──────────────────────────────────┘
python
# starlette/middleware/exceptions.py 核心逻辑（简化版）
class ExceptionHandlerMiddleware:
    async def __call__(self, scope, receive, send):
        try:
            await self.app(scope, receive, send)
        except Exception as exc:
            handler = self._lookup_handler(exc)
            if handler:
                response = await handler(request, exc)
                await response(scope, receive, send)
            else:
                raise exc  # 继续向上抛

    def _lookup_handler(self, exc: Exception) -> Callable | None:
        # 先查精确匹配
        if type(exc) in self.exception_handlers:
            return self.exception_handlers[type(exc)]
        # 再查 MRO（方法解析顺序）中的父类
        for cls in type(exc).__mro__:
            if cls in self.exception_handlers:
                return self.exception_handlers[cls]
        return None

5.2 HTTPException vs ValidationError ​

对比维度HTTPExceptionRequestValidationError
触发时机开发者主动 raisePydantic 参数校验失败自动抛出
状态码开发者自定义固定 422 Unprocessable Entity
响应格式{"detail": "..."}{"detail": [{"loc": [...], "msg": "...", "type": "..."}]}
异常类型fastapi.HTTPExceptionfastapi.exceptions.RequestValidationError
处理器注册@app.exception_handler(HTTPException)@app.exception_handler(RequestValidationError)
拦截点路由函数内部路由函数执行前（FastAPI 依赖注入阶段）
python
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

# 自定义 422 响应格式（简化 detail 结构）
@app.exception_handler(RequestValidationError)
async def custom_validation_handler(
    request: Request,
    exc: RequestValidationError,
) -> JSONResponse:
    simplified_errors: list[dict[str, str]] = []
    for error in exc.errors():
        simplified_errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
        })
    return JSONResponse(
        status_code=422,
        content={"error": "Validation Failed", "errors": simplified_errors},
    )

5.3 全局异常处理器的注册机制 ​

FastAPI 提供三层异常处理器注册：

python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from starlette.responses import PlainTextResponse

app: FastAPI = FastAPI()

# 层级 1：自定义异常 —— 精确匹配
@app.exception_handler(BusinessException)
async def handle_business(request: Request, exc: BusinessException) -> JSONResponse:
    return JSONResponse(status_code=exc.code, content={"error": exc.message})

# 层级 2：框架异常 —— 覆盖默认行为
@app.exception_handler(RequestValidationError)
async def handle_validation(request: Request, exc: RequestValidationError) -> JSONResponse:
    return JSONResponse(status_code=422, content={"error": "Invalid input"})

# 层级 3：全局兜底 —— 捕获所有未处理异常
@app.exception_handler(Exception)
async def handle_global(request: Request, exc: Exception) -> JSONResponse:
    # 生产环境应记录日志而非暴露异常细节
    import logging
    logging.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(status_code=500, content={"error": "Internal Server Error"})

注册机制内部流程：

@app.exception_handler(SomeException)
    │
    ▼
app.exception_handlers[SomeException] = handler_fn
    │
    ▼
Starlette 创建 ExceptionHandlerMiddleware
    │
    ├── exception_handlers: dict[type, Callable]
    └── 使用场景：__call__ 捕获异常后查表
注册方式作用域优先级
@app.exception_handler(CustomException)应用级高（精确匹配）
@app.exception_handler(HTTPException)应用级中（覆盖默认）
@app.exception_handler(Exception)应用级低（兜底）
APIRouter(exception_handler=...)路由级最高（局部覆盖）

5.4 性能考量 ​

操作耗时级别说明
HTTPException 抛出~0.001ms纯 Python 异常，无额外开销
异常处理器查表~0.001msdict 精确匹配 + MRO 线性查找
RequestValidationError~0.01msPydantic 已做校验，构造 error list 开销小
全局 Exception 处理器~0.001ms最后兜底，仅记录日志
异常栈生成~0.1msexc_info=True 时生成 traceback

5.5 设计动机 ​

设计选择动机
HTTPException 而非 HTTP 状态码直接返回统一异常抛出方式，集中由中间件处理响应格式
校验失败返回 422 而非 400422 语义更准确（请求格式正确但语义校验失败）
异常处理器注册机制解耦业务逻辑与错误响应格式，便于全局统一管理
MRO 继承链查找子类的异常若未注册处理器，自动回退到父类处理器
全局 Exception 兜底防止未捕获异常导致 500 默认 HTML 页面暴露内部信息

5.6 知识关联 ​

错误处理
├── HTTPException
│   ├── status_code: HTTP 状态码
│   ├── detail: 错误详情（string / dict / list）
│   └── headers: 额外响应头（如 WWW-Authenticate）
│
├── 异常处理链（Starlette）
│   ├── 精确匹配 type(exc)
│   ├── MRO 继承链查找父类
│   └── 未匹配则向上抛出
│
├── 校验错误（422）
│   ├── RequestValidationError：请求体/路径/查询参数校验失败
│   ├── WebSocketRequestValidationError：WebSocket 校验失败
│   └── 错误结构：[{"loc": ["body", "name"], "msg": "...", "type": "..."}]
│
├── 异常处理器注册
│   ├── @app.exception_handler(SomeException)
│   ├── APIRouter 局部处理器
│   └── 全局 Exception 兜底
│
└── 自定义异常模式
    ├── 继承 HTTPException（简单）
    ├── 继承 Exception + 独立处理器（业务解耦）
    └── BusinessException 统一错误码体系

总结 ​

知识点说明
HTTPExceptionHTTP 异常
自定义异常业务异常
异常处理器全局处理
错误响应统一格式