FastAPI异常处理完全指南

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

目录

FastAPI异常处理机制

FastAPI的异常体系

FastAPI继承了Starlette的异常处理机制，形成了完整的异常处理体系：

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

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

抛出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"
}

异常处理的核心原则

  1. 可预测性：客户端能预期错误响应格式
  2. 安全性：生产环境不暴露敏感信息
  3. 可追溯性：错误发生时能快速定位问题
  4. 一致性：所有API接口使用统一的错误格式

全局异常处理器

为什么需要全局处理？

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

注册全局异常处理器

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(),
        }
    )

异常处理器的执行顺序

# 1. 首先检查最具体的异常类型
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"code": 400, "message": f"值错误: {str(exc)}"}
    )

# 2. 然后是更一般的异常类型
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "未知错误"}
    )

# 注意：上面的general_exception_handler永远不会被触发
# 因为所有的ValueError都已经被第一个处理器处理了
# 所以要注意异常处理器的注册顺序

自定义业务异常类

创建统一的自定义异常

# exceptions.py — 统一的自定义异常
from typing import Optional, Dict, Any
from fastapi import HTTPException

class AppException(HTTPException):
    """基础业务异常"""
    def __init__(self, 
                 status_code: int = 400, 
                 message: str = "Bad Request", 
                 detail: Optional[str] = None,
                 headers: Optional[Dict[str, Any]] = None,
                 error_code: Optional[str] = None):
        super().__init__(
            status_code=status_code,
            detail=detail or message,
            headers=headers
        )
        self.message = message
        self.error_code = error_code

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

class UnauthorizedException(AppException):
    """未授权"""
    def __init__(self, reason: str = "请先登录"):
        super().__init__(
            status_code=401,
            message=reason,
            detail=reason,
            error_code="UNAUTHORIZED",
            headers={"WWW-Authenticate": "Bearer"}
        )

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

class RateLimitException(AppException):
    """请求过于频繁"""
    def __init__(self, retry_after: int = 60, reason: str = "请求过于频繁，请稍后重试"):
        super().__init__(
            status_code=429,
            message=reason,
            detail=reason,
            error_code="RATE_LIMIT_EXCEEDED"
        )
        self.retry_after = retry_after

class ValidationException(AppException):
    """参数验证异常"""
    def __init__(self, field: str, message: str):
        super().__init__(
            status_code=422,
            message=f"参数验证失败: {field}",
            detail=message,
            error_code="VALIDATION_ERROR"
        )

注册业务异常处理器

# 在应用中注册自定义异常处理器
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    content = {
        "code": exc.status_code,
        "message": exc.message,
        "error_code": exc.error_code,
        "path": str(request.url.path),
        "timestamp": datetime.utcnow().isoformat()
    }

    if hasattr(exc, 'retry_after'):
        content["retry_after"] = exc.retry_after

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

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

    return response

在业务逻辑中使用自定义异常

from exceptions import NotFoundException, UnauthorizedException, ValidationException

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    if user_id <= 0:
        raise ValidationException("user_id", "用户ID必须为正整数")
    
    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, current_user: dict = Depends(get_current_user)):
    article = await db.find_article(article_id)
    if not article:
        raise NotFoundException("文章", str(article_id))
    
    if article.author_id != current_user["id"]:
        raise UnauthorizedException("无权删除他人文章")
    
    await db.delete_article(article_id)
    return {"deleted": article_id, "message": "删除成功"}

统一响应格式

响应模型封装

from pydantic import BaseModel
from typing import TypeVar, Generic, Optional, Any, List, Dict
from datetime import datetime

T = TypeVar("T")

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

    class Config:
        json_schema_extra = {
            "example": {
                "code": 200,
                "message": "success",
                "data": {"id": 1, "name": "Alice"},
                "timestamp": "2024-01-15T10:30:00Z"
            }
        }

class ApiError(BaseModel):
    """错误响应格式"""
    code: int
    message: str
    error_code: Optional[str] = None
    path: str
    timestamp: str = datetime.utcnow().isoformat()
    details: Optional[Dict[str, Any]] = None

def success_response(data: Any = None, message: str = "success", path: str = None) -> ApiResponse:
    """成功响应"""
    return ApiResponse(
        code=200,
        message=message,
        data=data,
        path=path
    )

def error_response(code: int, 
                  message: str, 
                  error_code: str = None, 
                  path: str = None,
                  details: Dict[str, Any] = None) -> ApiError:
    """错误响应"""
    return ApiError(
        code=code,
        message=message,
        error_code=error_code,
        path=path,
        details=details
    )

在路由中使用统一响应格式

from fastapi import APIRouter
from typing import List

router = APIRouter()

@router.get("/users", response_model=ApiResponse[List[dict]])
async def list_users():
    try:
        users = await db.query_users()
        return success_response(data=users, message="获取用户列表成功")
    except Exception as e:
        logger.error(f"Failed to query users: {e}")
        return error_response(500, "获取用户列表失败")

@router.get("/users/{user_id}", response_model=ApiResponse[dict])
async def get_user(user_id: int):
    try:
        user = await db.find_user(user_id)
        if not user:
            return error_response(404, "用户不存在", "USER_NOT_FOUND")
        
        return success_response(data=user, message="获取用户成功")
    except ValueError:
        return error_response(400, "用户ID格式错误", "INVALID_USER_ID")
    except Exception as e:
        logger.error(f"Failed to get user {user_id}: {e}")
        return error_response(500, "获取用户失败", "INTERNAL_ERROR")

异常处理与日志

结构化日志配置

import structlog
import logging
from pythonjsonlogger import jsonlogger
from datetime import datetime

# 配置结构化日志
def configure_logging():
    # JSON格式的日志
    json_formatter = jsonlogger.JsonFormatter(
        '%(asctime)s %(name)s %(levelname)s %(message)s'
    )
    
    handler = logging.StreamHandler()
    handler.setFormatter(json_formatter)
    
    logger = logging.getLogger()
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    
    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.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            structlog.processors.JSONRenderer(),
        ],
        wrapper_class=structlog.stdlib.BoundLogger,
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        cache_logger_on_first_use=True,
    )

configure_logging()
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),
        query_params=dict(request.query_params),
        error_type=type(exc).__name__,
        error_message=str(exc),
        user_agent=request.headers.get("user-agent"),
        ip_address=request.client.host,
        traceback=traceback.format_exc() if logger.isEnabledFor(logging.DEBUG) else None,
    )

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

按异常类型差异化处理

@app.exception_handler(Exception)
async def smart_exception_handler(request: Request, exc: Exception):
    if isinstance(exc, AppException):
        # 业务异常：直接返回业务定义的错误
        logger.info(
            "business_exception",
            error_code=exc.error_code,
            message=exc.message,
            path=str(request.url.path)
        )
        return JSONResponse(
            status_code=exc.status_code,
            content={
                "code": exc.status_code, 
                "message": exc.message,
                "error_code": exc.error_code,
                "timestamp": datetime.utcnow().isoformat()
            }
        )

    if isinstance(exc, ValueError):
        # 业务参数错误 → 400
        logger.warning(
            "value_error",
            error=str(exc),
            path=str(request.url.path)
        )
        return JSONResponse(
            status_code=400, 
            content={
                "code": 400, 
                "message": str(exc),
                "error_code": "VALUE_ERROR",
                "timestamp": datetime.utcnow().isoformat()
            }
        )

    if isinstance(exc, ConnectionError):
        # 服务不可用 → 503
        logger.error("service_unavailable", error=str(exc))
        return JSONResponse(
            status_code=503, 
            content={
                "code": 503, 
                "message": "服务暂时不可用",
                "error_code": "SERVICE_UNAVAILABLE",
                "timestamp": datetime.utcnow().isoformat()
            }
        )

    # 未知异常 → 500 + 记录
    logger.exception(
        "unhandled_error", 
        error=str(exc), 
        path=str(request.url.path),
        traceback=traceback.format_exc()
    )
    return JSONResponse(
        status_code=500, 
        content={
            "code": 500, 
            "message": "服务器错误",
            "error_code": "INTERNAL_ERROR",
            "timestamp": datetime.utcnow().isoformat()
        }
    )

404错误处理

全局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:
        logger.info(
            "not_found",
            path=str(request.url.path),
            method=request.method
        )
        return JSONResponse(
            status_code=404,
            content={
                "code": 404,
                "message": f"API 路径 '{request.url.path}' 不存在",
                "suggestion": "请检查 URL 是否正确，或访问 /docs 查看可用接口",
                "timestamp": datetime.utcnow().isoformat()
            }
        )
    # 其他 HTTP 异常走通用处理
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code, 
            "message": exc.detail,
            "timestamp": datetime.utcnow().isoformat()
        }
    )

自定义404响应格式

# 更详细的404处理
@app.exception_handler(StarletteHTTPException)
async def detailed_404_handler(request: Request, exc: StarletteHTTPException):
    if exc.status_code == 404:
        # 检查是否是API路径
        if request.url.path.startswith("/api/"):
            suggestion = "请检查API版本或端点路径是否正确"
        else:
            suggestion = "请检查URL路径是否正确"
            
        return JSONResponse(
            status_code=404,
            content={
                "code": 404,
                "message": f"请求的资源不存在",
                "requested_path": request.url.path,
                "method": request.method,
                "timestamp": datetime.utcnow().isoformat(),
                "suggestion": suggestion,
                "available_endpoints": [
                    "/api/v1/users",
                    "/api/v1/products",
                    "/api/v1/orders"
                ]
            }
        )
    # 其他HTTP异常处理
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code,
            "message": exc.detail,
            "timestamp": datetime.utcnow().isoformat()
        }
    )

验证错误详解

RequestValidationError结构

# 触发422验证错误的不同方式
@app.get("/calculate")
async def calculate(x: int, y: int):
    if y == 0:
        raise HTTPException(status_code=400, detail="除数不能为零")
    return {"result": x / y}

# GET /calculate?x=10&y=0 → 业务逻辑错误（HTTPException）
# GET /calculate?x=abc&y=2 → 参数类型验证错误（RequestValidationError）

验证错误响应示例：

{
  "code": 422,
  "message": "参数校验失败",
  "errors": [
    {
      "field": "query.x",
      "message": "Input should be a valid integer, unable to parse string as an integer"
    }
  ]
}

自定义验证错误消息

from pydantic import BaseModel, field_validator, ValidationError
from typing import Optional

class CreateUser(BaseModel):
    name: str
    email: str
    age: int
    phone: Optional[str] = None

    @field_validator("name")
    @classmethod
    def name_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("用户名不能为空")
        if len(v.strip()) < 2:
            raise ValueError("用户名长度至少为2个字符")
        return v.strip()

    @field_validator("email")
    @classmethod
    def email_format(cls, v: str) -> str:
        if "@" not in v or "." 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

    @field_validator("phone")
    @classmethod
    def phone_format(cls, v: Optional[str]) -> Optional[str]:
        if v is not None:
            # 简单的手机号验证
            import re
            if not re.match(r"^1[3-9]\d{9}$", v):
                raise ValueError("手机号格式不正确")
        return v

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

处理嵌套模型验证错误

class Address(BaseModel):
    street: str
    city: str
    zip_code: str

    @field_validator("zip_code")
    @classmethod
    def zip_code_format(cls, v: str) -> str:
        import re
        if not re.match(r"^\d{6}$", v):
            raise ValueError("邮编格式不正确，应为6位数字")
        return v

class UserWithAddress(BaseModel):
    name: str
    address: Address

# 错误处理示例
@app.exception_handler(RequestValidationError)
async def enhanced_validation_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        loc = error["loc"]
        field = " -> ".join(str(loc_part) for loc_part in loc)
        errors.append({
            "field": field,
            "message": error["msg"],
            "type": error["type"],
            "input": error.get("input")
        })

    return JSONResponse(
        status_code=422,
        content={
            "code": 422,
            "message": "参数校验失败",
            "validation_errors": errors,
            "timestamp": datetime.utcnow().isoformat()
        }
    )

异常处理最佳实践

推荐做法

✅ 推荐：
1. 自定义业务异常类，统一错误码和消息格式
2. 异常处理器返回统一的 JSON 结构
3. 生产环境不要暴露异常堆栈
4. 记录详细日志供排查
5. HTTPException 用于"预期内的错误"
6. 全局兜底异常处理器防止意外 500 泄漏
7. 区分客户端错误和服务器错误
8. 为不同类型的错误提供相应的错误码
9. 在异常中包含足够的上下文信息
10. 实现优雅降级机制

实现优雅的错误处理

# 完整的错误处理示例
from enum import Enum

class ErrorCode(str, Enum):
    """错误码枚举"""
    # 通用错误
    INTERNAL_ERROR = "INTERNAL_ERROR"
    VALIDATION_ERROR = "VALIDATION_ERROR"
    RESOURCE_NOT_FOUND = "RESOURCE_NOT_FOUND"
    UNAUTHORIZED = "UNAUTHORIZED"
    FORBIDDEN = "FORBIDDEN"
    RATE_LIMIT_EXCEEDED = "RATE_LIMIT_EXCEEDED"
    
    # 业务错误
    USER_ALREADY_EXISTS = "USER_ALREADY_EXISTS"
    INVALID_CREDENTIALS = "INVALID_CREDENTIALS"
    INSUFFICIENT_BALANCE = "INSUFFICIENT_BALANCE"

class DetailedApiError(BaseModel):
    """详细错误响应"""
    code: int
    error_code: ErrorCode
    message: str
    details: Optional[Dict[str, Any]] = None
    path: str
    timestamp: str = datetime.utcnow().isoformat()
    trace_id: Optional[str] = None  # 用于追踪错误

# 为每个异常类型创建专门的处理器
@app.exception_handler(StarletteHTTPException)
async def standard_http_exception_handler(request: Request, exc: StarletteHTTPException):
    error_map = {
        400: ErrorCode.VALIDATION_ERROR,
        401: ErrorCode.UNAUTHORIZED,
        403: ErrorCode.FORBIDDEN,
        404: ErrorCode.RESOURCE_NOT_FOUND,
        429: ErrorCode.RATE_LIMIT_EXCEEDED,
    }
    
    error_code = error_map.get(exc.status_code, ErrorCode.INTERNAL_ERROR)
    
    return JSONResponse(
        status_code=exc.status_code,
        content=DetailedApiError(
            code=exc.status_code,
            error_code=error_code,
            message=exc.detail,
            path=str(request.url.path),
            details={"original_status": exc.status_code}
        ).model_dump()
    )

@app.exception_handler(RequestValidationError)
async def detailed_validation_handler(request: Request, exc: RequestValidationError):
    validation_errors = []
    for error in exc.errors():
        validation_errors.append({
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })
    
    return JSONResponse(
        status_code=422,
        content=DetailedApiError(
            code=422,
            error_code=ErrorCode.VALIDATION_ERROR,
            message="请求参数验证失败",
            details={"errors": validation_errors},
            path=str(request.url.path)
        ).model_dump()
    )

性能优化建议

1. 异常处理的性能考虑

# 避免在正常业务流程中使用异常
# ❌ 错误：用异常控制流程
def get_user_safe(user_id: int):
    try:
        return db.get_user(user_id)
    except NotFoundError:
        return None

# ✅ 正确：先检查再操作
def get_user_optimized(user_id: int):
    if db.user_exists(user_id):
        return db.get_user(user_id)
    return None

2. 日志性能优化

import asyncio
from concurrent.futures import ThreadPoolExecutor

# 使用异步日志记录避免阻塞
class AsyncLogger:
    def __init__(self):
        self.executor = ThreadPoolExecutor(max_workers=2)
    
    async def log_error_async(self, message: str, **kwargs):
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(
            self.executor, 
            lambda: logger.error(message, **kwargs)
        )

async_logger = AsyncLogger()

@app.exception_handler(Exception)
async def async_logging_handler(request: Request, exc: Exception):
    # 异步记录日志，避免阻塞响应
    await async_logger.log_error_async(
        "unhandled_error",
        error=str(exc),
        path=str(request.url.path),
        traceback=traceback.format_exc()
    )
    
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器错误"}
    )

3. 异常缓存策略

from functools import lru_cache
import time

# 对于频繁发生的可预见错误，可以考虑缓存错误响应
class ErrorCache:
    def __init__(self, ttl: int = 60):
        self.cache = {}
        self.ttl = ttl
    
    def get(self, key: str):
        if key in self.cache:
            value, timestamp = self.cache[key]
            if time.time() - timestamp < self.ttl:
                return value
            del self.cache[key]
        return None
    
    def set(self, key: str, value):
        self.cache[key] = (value, time.time())

error_cache = ErrorCache()

@app.get("/users/{user_id}")
async def get_user_cached(user_id: int):
    cache_key = f"user_not_found_{user_id}"
    
    # 检查缓存的错误
    cached_error = error_cache.get(cache_key)
    if cached_error:
        return JSONResponse(status_code=404, content=cached_error)
    
    user = await db.find_user(user_id)
    if not user:
        error_response = {
            "code": 404,
            "message": f"用户 {user_id} 不存在",
            "timestamp": datetime.utcnow().isoformat()
        }
        # 缓存错误响应，避免频繁查询
        error_cache.set(cache_key, error_response)
        return JSONResponse(status_code=404, content=error_response)
    
    return user

常见陷阱与避坑指南

陷阱1：异常处理器顺序错误

# ❌ 错误：通用异常处理器放在前面
@app.exception_handler(Exception)
async def general_handler(request: Request, exc: Exception):
    return JSONResponse(500, {"error": "General error"})

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    # 这个处理器永远不会被执行
    return JSONResponse(400, {"error": "Value error"})

# ✅ 正确：具体异常处理器放在前面
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(400, {"error": "Value error"})

@app.exception_handler(Exception)
async def general_handler(request: Request, exc: Exception):
    return JSONResponse(500, {"error": "General error"})

陷阱2：在异常处理器中引发新异常

# ❌ 错误：异常处理器本身也抛出异常
@app.exception_handler(Exception)
async def bad_handler(request: Request, exc: Exception):
    # 如果这里的操作失败，会导致双重异常
    problematic_operation()
    return JSONResponse(500, {"error": "Handled"})

# ✅ 正确：异常处理器应该是安全的
@app.exception_handler(Exception)
async def safe_handler(request: Request, exc: Exception):
    try:
        # 安全的日志记录
        logger.error(f"Error occurred: {exc}")
    except:
        # 即使日志记录失败也不能影响响应
        pass
    return JSONResponse(500, {"error": "Internal server error"})

陷阱3：暴露敏感信息

# ❌ 错误：在生产环境中暴露详细错误信息
@app.exception_handler(Exception)
async def leaky_handler(request: Request, exc: Exception):
    return JSONResponse(
        500, 
        {
            "error": str(exc),
            "traceback": traceback.format_exc(),  # 泄露敏感信息
            "locals": locals()  # 泄露局部变量
        }
    )

# ✅ 正确：生产环境隐藏敏感信息
@app.exception_handler(Exception)
async def secure_handler(request: Request, exc: Exception):
    # 只在开发环境记录详细信息
    if os.getenv("ENVIRONMENT") == "development":
        logger.error(f"Error details: {exc}\n{traceback.format_exc()}")
    else:
        logger.error(f"Error occurred: {type(exc).__name__}")
    
    return JSONResponse(
        500, 
        {"error": "Internal server error"}
    )

陷阱4：忽略异常上下文

# ❌ 错误：丢失异常上下文
@app.exception_handler(SomeException)
async def poor_context_handler(request: Request, exc: SomeException):
    # 没有保留原始异常的上下文信息
    return JSONResponse(500, {"error": "Something went wrong"})

# ✅ 正确：保留有用的上下文信息
@app.exception_handler(SomeException)
async def rich_context_handler(request: Request, exc: SomeException):
    logger.error(
        "Business exception occurred",
        path=str(request.url.path),
        method=request.method,
        user_id=getattr(request.state, 'user_id', 'unknown'),
        error=str(exc)
    )
    return JSONResponse(
        500, 
        {
            "error": "Operation failed",
            "code": "BUSINESS_ERROR",
            "timestamp": datetime.utcnow().isoformat()
        }
    )

相关教程

良好的异常处理是构建可靠API的关键。建议为每种业务场景定义专门的异常类型，并实现统一的错误响应格式。同时，确保在生产环境中不会泄露敏感信息，并记录足够的日志以便调试。 异常处理虽然重要，但不应影响API的性能。避免在正常业务流程中使用异常控制流程，合理使用日志异步处理，并对可预见的错误进行缓存优化。

总结

异常类型触发场景推荐处理方式
HTTPException业务逻辑（资源不存在、权限不足）使用自定义业务异常类
RequestValidationError参数校验失败统一验证错误处理器
Exception所有未捕获异常全局兜底异常处理器
自定义 AppException业务特定错误分类处理，统一格式

💡 核心原则：异常处理的目标是让客户端收到可预期的错误响应，同时让服务端记录足够排查问题的日志，并在生产环境中保护敏感信息不被泄露。

FastAPI的异常处理机制为企业级应用提供了强大的错误处理能力，通过合理的异常分类、统一的响应格式和完善的日志记录，可以构建出稳定可靠的API服务。

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业务特定错误自定义结构

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

🔗 扩展阅读