error-handling

错误处理

遵循 让它崩溃 理念，为 Python API 实现可用于生产环境的错误处理。

设计理念

让它崩溃——不要防御式编程。让异常自然传播，并在边界处处理。

# BAD - Too defensive, obscures errors
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    try:
        user = await user_service.get(user_id)
        if not user:
            raise HTTPException(404, "Not found")
        return user
    except DatabaseError as e:
        raise HTTPException(500, "Database error")
    except Exception as e:
        logger.exception("Unexpected error")
        raise HTTPException(500, "Internal error")

# GOOD - Let exceptions propagate, handle at boundary
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

核心原则

1. 底层抛出，高层捕获——在错误发生处抛出异常，在 API 边界处处理
2. 领域异常——创建有语义的异常，而非通用异常
3. 全局处理器——使用 @app.exception_handler() 集中格式化错误
4. 不使用裸 except——始终捕获特定异常
5. 保留上下文——使用 raise ... from error 保留原始回溯信息

快速开始

1. 定义领域异常

from enum import StrEnum

class ErrorCode(StrEnum):
    USER_NOT_FOUND = "user_not_found"
    INVALID_CREDENTIALS = "invalid_credentials"
    RATE_LIMITED = "rate_limited"

class DomainError(Exception):
    """Base exception for all domain errors."""
    def __init__(self, code: ErrorCode, message: str, status_code: int = 400):
        self.code = code
        self.message = message
        self.status_code = status_code
        super().__init__(message)

class UserNotFoundError(DomainError):
    def __init__(self, user_id: int):
        super().__init__(
            code=ErrorCode.USER_NOT_FOUND,
            message=f"User {user_id} not found",
            status_code=404
        )

2. 定义错误响应架构

from pydantic import BaseModel

class ErrorDetail(BaseModel):
    code: str
    message: str
    request_id: str | None = None

class ErrorResponse(BaseModel):
    error: ErrorDetail

3. 注册全局处理器

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

app = FastAPI()

@app.exception_handler(DomainError)
async def domain_error_handler(request: Request, exc: DomainError):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": exc.code, "message": exc.message}}
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": {"code": "http_error", "message": str(exc.detail)}}
    )

@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=422,
        content={"error": {"code": "validation_error", "message": "Invalid request"}}
    )

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception):
    # Log full error internally
    logger.exception("Unhandled error")
    # Return safe message to client
    return JSONResponse(
        status_code=500,
        content={"error": {"code": "internal_error", "message": "Internal server error"}}
    )

4. 在路由中使用

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = await user_service.get(user_id)
    if not user:
        raise UserNotFoundError(user_id)
    return user

何时捕获异常

仅在以下情况下捕获异常：

情况	示例
需要重试	使用 tenacity.retry() 处理暂时性故障
需要转换	将第三方 SDK 错误包装为领域错误
需要清理	使用 finally 或上下文管理器
需要添加上下文	raise DomainError(...) from original

Python + FastAPI 集成

层	职责
服务/领域	抛出领域异常（UserNotFoundError）
路由	让异常传播（不使用 try/except）
异常处理器	转换为 HTTP 响应
中间件	添加请求上下文（request_id、计时信息）

常见模式

包装第三方 SDK

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

class ExternalServiceError(DomainError):
    def __init__(self, service: str, original: Exception):
        super().__init__(
            code=ErrorCode.EXTERNAL_SERVICE_ERROR,
            message=f"{service} unavailable",
            status_code=503
        )
        self.__cause__ = original

@retry(stop=stop_after_attempt(3), wait=wait_exponential())
async def call_payment_api(data: dict):
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.post("https://api.payment.com/charge", json=data)
            response.raise_for_status()
            return response.json()
    except httpx.HTTPError as e:
        raise ExternalServiceError("Payment API", e) from e

后台任务错误处理

from fastapi import BackgroundTasks

async def safe_background_task(task_func, *args, **kwargs):
    try:
        await task_func(*args, **kwargs)
    except Exception as e:
        logger.exception(f"Background task failed: {e}")
        # Optional: send to dead letter queue or alerting

@app.post("/orders")
async def create_order(order: Order, background_tasks: BackgroundTasks):
    result = await order_service.create(order)
    background_tasks.add_task(safe_background_task, send_confirmation_email, result.id)
    return result

故障排查

问题	原因	修复方法
响应中出现堆栈跟踪	缺少通用处理器	添加 @app.exception_handler(Exception)
原始错误丢失	缺少 from	使用 raise NewError() from original
验证错误泄露	默认处理器	覆盖 RequestValidationError 处理器
故障被静默处理	异常被吞掉	让异常传播，并在边界处处理

参考资料

• Python 模式 - 异常设计、何时捕获、SDK 包装
• FastAPI 模式 - HTTPException、全局处理器、中间件
• Pydantic 模式 - ValidationError、在验证器中抛出异常
• Asyncio 模式 - TaskGroup、超时、后台任务
• FastAPI 文档：处理错误
• Pydantic 文档：错误处理