FastAPI 异常处理
概述
良好的异常处理是构建健壮Web应用的关键。FastAPI提供了强大的异常处理机制，支持HTTP异常、自定义异常处理器、全局错误处理等。本章将探讨如何在FastAPI中优雅地处理各种异常情况。
🚨 HTTP异常基础
基础HTTP异常
python
from fastapi import FastAPI, HTTPException, status
from fastapi.responses import JSONResponse
import logging
app = FastAPI()
logger = logging.getLogger(__name__)
# 模拟数据库
fake_users = {
1: {"id": 1, "name": "Alice", "email": "alice@example.com"},
2: {"id": 2, "name": "Bob", "email": "bob@example.com"}
}
@app.get("/users/{user_id}")
async def get_user(user_id: int):
# 参数验证
if user_id <= 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="用户ID必须大于0"
)
# 查找用户
user = fake_users.get(user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"用户ID {user_id} 不存在"
)
return user
@app.post("/users/")
async def create_user(user_data: dict):
# 业务逻辑验证
if "name" not in user_data or not user_data["name"].strip():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="用户名称不能为空"
)
# 检查邮箱重复
email = user_data.get("email")
for existing_user in fake_users.values():
if existing_user["email"] == email:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="邮箱地址已存在"
)
# 创建用户
new_id = max(fake_users.keys()) + 1 if fake_users else 1
new_user = {"id": new_id, "name": user_data["name"], "email": email}
fake_users[new_id] = new_user
return new_user带详细信息的异常
python
from typing import Any, Dict, Optional
from pydantic import BaseModel
class DetailedHTTPException(HTTPException):
def __init__(
self,
status_code: int,
detail: str,
error_code: str,
field: Optional[str] = None,
value: Optional[Any] = None
):
error_detail = {
"error_code": error_code,
"message": detail,
"field": field,
"value": value
}
super().__init__(status_code=status_code, detail=error_detail)
@app.post("/products/")
async def create_product(product_data: dict):
if "price" not in product_data:
raise DetailedHTTPException(
status_code=422,
detail="产品价格是必需的",
error_code="PRICE_REQUIRED",
field="price"
)
try:
price = float(product_data["price"])
if price <= 0:
raise DetailedHTTPException(
status_code=422,
detail="产品价格必须大于0",
error_code="INVALID_PRICE",
field="price",
value=price
)
except ValueError:
raise DetailedHTTPException(
status_code=422,
detail="产品价格必须是有效的数字",
error_code="PRICE_FORMAT_ERROR",
field="price",
value=product_data["price"]
)
return {"message": "产品创建成功", "product": product_data}🎯 自定义异常类
业务异常类
python
class BusinessException(Exception):
"""业务异常基类"""
def __init__(self, message: str, error_code: str, status_code: int = 400):
self.message = message
self.error_code = error_code
self.status_code = status_code
super().__init__(self.message)
class UserNotFoundError(BusinessException):
def __init__(self, user_id: int):
super().__init__(
message=f"用户 {user_id} 不存在",
error_code="USER_NOT_FOUND",
status_code=404
)
self.user_id = user_id
class DuplicateEmailError(BusinessException):
def __init__(self, email: str):
super().__init__(
message=f"邮箱 {email} 已被使用",
error_code="DUPLICATE_EMAIL",
status_code=409
)
self.email = email
class InvalidInputError(BusinessException):
def __init__(self, field: str, value: Any, reason: str):
super().__init__(
message=f"字段 {field} 的值无效: {reason}",
error_code="INVALID_INPUT",
status_code=422
)
self.field = field
self.value = value
# 使用自定义异常
@app.get("/api/users/{user_id}")
async def get_user_api(user_id: int):
if user_id not in fake_users:
raise UserNotFoundError(user_id)
return fake_users[user_id]
@app.post("/api/users/")
async def create_user_api(user_data: dict):
# 验证输入
if not user_data.get("name", "").strip():
raise InvalidInputError("name", user_data.get("name"), "名称不能为空")
email = user_data.get("email", "")
if not email or "@" not in email:
raise InvalidInputError("email", email, "邮箱格式无效")
# 检查重复邮箱
for user in fake_users.values():
if user["email"] == email:
raise DuplicateEmailError(email)
# 创建用户
new_id = max(fake_users.keys()) + 1 if fake_users else 1
new_user = {"id": new_id, "name": user_data["name"], "email": email}
fake_users[new_id] = new_user
return new_user🛠️ 异常处理器
全局异常处理器
python
from fastapi import Request
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import traceback
from datetime import datetime
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
"""HTTP异常处理器"""
logger.error(f"HTTP异常: {exc.status_code} - {exc.detail}")
return JSONResponse(
status_code=exc.status_code,
content={
"error": {
"type": "http_error",
"status_code": exc.status_code,
"message": exc.detail,
"path": str(request.url.path),
"timestamp": datetime.now().isoformat()
}
}
)
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""请求验证异常处理器"""
return JSONResponse(
status_code=422,
content={
"error": {
"type": "validation_error",
"message": "请求数据验证失败",
"details": exc.errors(),
"path": str(request.url.path),
"timestamp": datetime.now().isoformat()
}
}
)
@app.exception_handler(BusinessException)
async def business_exception_handler(request: Request, exc: BusinessException):
"""业务异常处理器"""
logger.warning(f"业务异常: {exc.error_code} - {exc.message}")
error_response = {
"error": {
"type": "business_error",
"error_code": exc.error_code,
"message": exc.message,
"path": str(request.url.path),
"timestamp": datetime.now().isoformat()
}
}
# 添加特定异常的额外信息
if isinstance(exc, UserNotFoundError):
error_response["error"]["user_id"] = exc.user_id
elif isinstance(exc, DuplicateEmailError):
error_response["error"]["email"] = exc.email
elif isinstance(exc, InvalidInputError):
error_response["error"]["field"] = exc.field
error_response["error"]["value"] = exc.value
return JSONResponse(status_code=exc.status_code, content=error_response)
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
"""通用异常处理器"""
logger.error(f"未处理的异常: {str(exc)}", exc_info=True)
error_response = {
"error": {
"type": "internal_error",
"message": "服务器内部错误",
"path": str(request.url.path),
"timestamp": datetime.now().isoformat()
}
}
# 开发环境显示详细错误信息
import os
if os.getenv("ENVIRONMENT") == "development":
error_response["error"]["details"] = str(exc)
error_response["error"]["traceback"] = traceback.format_exc()
return JSONResponse(status_code=500, content=error_response)🔍 异常监控和重试
错误统计
python
from collections import defaultdict
class ErrorMonitor:
def __init__(self):
self.error_counts = defaultdict(int)
def record_error(self, error_type: str, error_code: str = None):
key = f"{error_type}:{error_code}" if error_code else error_type
self.error_counts[key] += 1
def get_stats(self) -> dict:
return {
"error_counts": dict(self.error_counts),
"total_errors": sum(self.error_counts.values())
}
error_monitor = ErrorMonitor()
@app.exception_handler(BusinessException)
async def monitored_business_exception_handler(request: Request, exc: BusinessException):
error_monitor.record_error("business_error", exc.error_code)
return JSONResponse(
status_code=exc.status_code,
content={
"error": {
"type": "business_error",
"error_code": exc.error_code,
"message": exc.message,
"timestamp": datetime.now().isoformat()
}
}
)
@app.get("/admin/error-stats/")
async def get_error_statistics():
return error_monitor.get_stats()重试机制
python
import asyncio
from functools import wraps
def retry_on_exception(max_retries: int = 3, delay: float = 1.0):
"""重试装饰器"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except Exception as e:
last_exception = e
if attempt == max_retries:
break
await asyncio.sleep(delay * (2 ** attempt))
raise last_exception
return wrapper
return decorator
@retry_on_exception(max_retries=3, delay=1.0)
async def unreliable_operation():
"""不可靠的操作"""
import random
if random.random() < 0.7: # 70%失败率
raise Exception("操作失败")
return {"result": "成功"}
@app.get("/unreliable/")
async def call_unreliable_operation():
try:
result = await unreliable_operation()
return result
except Exception:
return {"result": "降级响应", "message": "使用缓存数据"}总结
FastAPI的异常处理机制包括：
- ✅ HTTP异常：基础异常和详细异常信息
- ✅ 自定义异常：业务异常类和验证异常
- ✅ 异常处理器：全局和特定异常处理
- ✅ 监控统计：错误记录和统计分析
- ✅ 错误恢复：重试机制和降级策略
良好的异常处理能提升用户体验并帮助开发者快速定位问题。
异常处理最佳实践
- 使用具体的HTTP状态码
- 提供有意义的错误消息
- 记录详细的错误日志
- 实现适当的错误监控
- 考虑降级和重试策略
在下一章中，我们将学习FastAPI的项目结构组织和模块化设计。