1. FastAPI异常处理基础

1.1 异常的基本概念与作用

异常是程序运行过程中发生的特殊情况,用于表示程序执行过程中的错误或异常状态。在FastAPI中,异常处理是确保API能够优雅地处理错误情况并返回适当响应的重要机制。

1.2 内置异常类型与默认行为

FastAPI提供了多种内置异常类型,每种类型都有其特定的用途和默认行为:

异常类型 描述 默认行为
HTTPException 用于返回HTTP错误响应 根据状态码返回对应的HTTP错误响应
RequestValidationError 请求参数验证失败时抛出 返回422 Unprocessable Entity错误
ValidationError

Pydantic模型验证失败时抛出

通常由FastAPI自动捕获并转换为RequestValidationError
WebSocketDisconnect WebSocket连接断开时抛出 关闭WebSocket连接

1.3 异常响应的标准结构

FastAPI的异常响应通常具有以下标准结构:

{
  "detail": "错误信息"
}

对于验证错误,响应结构会更加详细:

{
  "detail": [
    {
      "loc": ["body", "field_name"],
      "msg": "错误信息",
      "type": "错误类型"
    }
  ]
}

1.4 异常处理的基本原则

  1. 早发现早处理:在错误发生的最早阶段捕获并处理异常
  2. 统一处理:使用全局异常处理器统一处理相同类型的异常
  3. 提供清晰的错误信息:确保错误信息对客户端友好且有帮助
  4. 记录异常:记录异常信息以便于调试和监控
  5. 避免暴露敏感信息:不在错误响应中包含敏感信息

2. 自定义异常实现

2.1 自定义异常类的创建与继承

在FastAPI中,我们可以通过继承Exception类来创建自定义异常:

# 自定义异常类
class CustomException(Exception):
    def __init__(self, message: str, status_code: int = 400):
        self.message = message
        self.status_code = status_code
        super().__init__(self.message)

# 继承自HTTPException
from fastapi import HTTPException

class NotFoundException(HTTPException):
    def __init__(self, resource: str, resource_id: str):
        super().__init__(
            status_code=404,
            detail=f"{resource} with id {resource_id} not found"
        )

2.2 异常处理装饰器的参数与使用方法

FastAPI提供了@app.exception_handler()装饰器来注册异常处理器,其参数包括:

  • exc_type:要处理的异常类型
  • 处理函数:接收requestexc两个参数,返回响应
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 注册自定义异常处理器
@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )

2.3 异常响应模型的自定义配置

我们可以使用Pydantic模型来定义异常响应的结构:

from pydantic import BaseModel

class ErrorResponse(BaseModel):
    detail: str
    error_code: int
    timestamp: str

@app.exception_handler(CustomException)
async def custom_exception_handler(request: Request, exc: CustomException):
    from datetime import datetime
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(
            detail=exc.message,
            error_code=exc.status_code,
            timestamp=datetime.now().isoformat()
        ).dict()
    )

2.4 异常的国际化处理

对于多语言应用,我们可以使用国际化库来处理异常信息:

from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette_context import context

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # 获取当前语言
    language = context.get("language", "en")
    # 根据语言获取错误信息
    error_messages = get_translated_error_messages(exc, language)
    return JSONResponse(
        status_code=422,
        content={"detail": error_messages}
    )

2.5 具体场景的异常实现示例

2.5.1 业务逻辑异常
class InsufficientFundsException(Exception):
    def __init__(self, balance: float, required: float):
        self.message = f"Insufficient funds: balance={balance}, required={required}"
        self.status_code = 400
        super().__init__(self.message)

@app.exception_handler(InsufficientFundsException)
async def insufficient_funds_handler(request: Request, exc: InsufficientFundsException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )

@app.post("/transfer")
async def transfer(from_account: int, to_account: int, amount: float):
    # 检查余额
    balance = get_account_balance(from_account)
    if balance < amount:
        raise InsufficientFundsException(balance, amount)
    # 执行转账
    # ...
    return {"message": "Transfer successful"}
2.5.2 权限异常
class PermissionDeniedException(Exception):
    def __init__(self, action: str, resource: str):
        self.message = f"Permission denied: cannot {action} {resource}"
        self.status_code = 403
        super().__init__(self.message)

@app.exception_handler(PermissionDeniedException)
async def permission_denied_handler(request: Request, exc: PermissionDeniedException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )

@app.get("/admin/users")
async def get_admin_users(current_user: User = Depends(get_current_user)):
    if not current_user.is_admin:
        raise PermissionDeniedException("access", "admin users")
    # 获取管理员用户列表
    # ...
    return {"users": admin_users}

3. 全局异常处理器

3.1 @app.exception_handler() 

@app.exception_handler()装饰器接受以下参数:

  • exc_type:要处理的异常类型,可以是内置异常或自定义异常
  • 处理函数:一个异步函数,接收requestexc两个参数,返回响应对象

3.2 不同类型异常的统一处理策略

我们可以为不同类型的异常注册不同的处理器,也可以为所有异常注册一个统一的处理器:

# 为所有异常注册统一处理器
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 记录异常
    logger.error(f"Unhandled exception: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

# 为特定异常注册处理器
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(
        status_code=400,
        content={"detail": str(exc)}
    )

3.3 全局异常处理的注册顺序与优先级

FastAPI会按照异常处理器的注册顺序来匹配异常类型,具体规则如下:

  1. 首先尝试匹配最具体的异常类型
  2. 如果没有找到匹配的处理器,会尝试匹配父类异常类型
  3. 最后会使用默认的异常处理器

因此,我们应该先注册具体异常的处理器,再注册通用异常的处理器:

# 正确的注册顺序
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    # 处理ValueError
    pass

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 处理所有其他异常
    pass

3.4 异常日志记录与上下文信息

在异常处理器中,我们应该记录异常信息以便于调试和监控:

import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 记录异常,包含请求信息
    logger.error(
        f"Unhandled exception: {exc}",
        exc_info=True,
        extra={
            "path": request.url.path,
            "method": request.method,
            "query_params": dict(request.query_params)
        }
    )
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

3.5 异常处理与中间件的关系

异常处理器和中间件都可以用于处理异常,但它们的职责不同:

  • 中间件:用于处理请求和响应的通用逻辑,包括异常的捕获和处理
  • 异常处理器:专门用于处理特定类型的异常,提供更详细的错误响应

我们可以在中间件中捕获异常并进行预处理,然后让异常处理器处理具体的异常:

from fastapi import FastAPI, Request
from fastapi.middleware.base import BaseHTTPMiddleware

app = FastAPI()

class ErrorHandlingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        try:
            response = await call_next(request)
            return response
        except Exception as exc:
            # 记录异常
            logger.error(f"Exception in middleware: {exc}", exc_info=True)
            # 重新抛出异常,让异常处理器处理
            raise

app.add_middleware(ErrorHandlingMiddleware)

4. HTTP异常处理

4.1 HTTPException 类的详细参数与使用

HTTPException是FastAPI中用于返回HTTP错误响应的内置异常类,其参数包括:

  • status_code:HTTP状态码(必需)
  • detail:错误详情(可选)
  • headers:响应头(可选)
from fastapi import HTTPException

# 基本用法
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=404,
            detail=f"Item {item_id} not found"
        )
    return {"item": items[item_id]}

# 使用自定义响应头
@app.get("/secret")
async def get_secret():
    raise HTTPException(
        status_code=401,
        detail="Not authenticated",
        headers={"WWW-Authenticate": "Bearer"},
    )

4.2 状态码与异常的对应关系

状态码 含义 常见使用场景
400 Bad Request 请求参数错误
401 Unauthorized 未授权访问
403 Forbidden 权限不足
404 Not Found 资源不存在
405 Method Not Allowed 请求方法不允许
422 Unprocessable Entity 请求参数验证失败
500 Internal Server Error 服务器内部错误
503 Service Unavailable 服务不可用

4.3 自定义HTTP异常的实现

我们可以通过继承HTTPException来创建自定义HTTP异常:

from fastapi import HTTPException

class NotFoundException(HTTPException):
    def __init__(self, resource: str, resource_id: str):
        super().__init__(
            status_code=404,
            detail=f"{resource} with id {resource_id} not found"
        )

class BadRequestException(HTTPException):
    def __init__(self, detail: str):
        super().__init__(
            status_code=400,
            detail=detail
        )

# 使用自定义HTTP异常
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)
    if not user:
        raise NotFoundException("User", str(user_id))
    return user

4.4 常见HTTP错误场景的处理

4.4.1 场景1:资源不存在
@app.get("/items/{item_id}")
async def read_item(item_id: int):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(
            status_code=404,
            detail=f"Item {item_id} not found"
        )
    return item
4.4.2 场景2:权限不足
from fastapi import Depends, HTTPException
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if not current_user.is_active:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user
4.4.3 场景3:请求参数错误
@app.post("/items")
async def create_item(item: ItemCreate):
    if item.price <= 0:
        raise HTTPException(
            status_code=400,
            detail="Price must be greater than 0"
        )
    # 创建物品
    # ...
    return item

4.5 HTTP异常的最佳实践

  1. 使用具体的状态码:根据错误类型选择合适的HTTP状态码
  2. 提供清晰的错误信息:确保错误信息对客户端友好且有帮助
  3. 使用自定义HTTP异常:为常见错误创建自定义HTTP异常类,提高代码可读性
  4. 记录异常信息:在抛出异常前记录相关信息,便于调试和监控
  5. 避免暴露敏感信息:不在错误响应中包含敏感信息

5. 验证异常处理

5.1 Pydantic验证异常的详细参数

Pydantic的ValidationError包含以下主要参数:

  • errors:错误列表,每个错误包含以下字段:
    • loc:错误位置(如[“body”, “field_name”])
    • msg:错误信息
    • type:错误类型
    • ctx:错误上下文

5.2 请求参数验证失败的响应定制

我们可以通过注册RequestValidationError处理器来自定义验证错误的响应:

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

app = FastAPI()

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

5.3 响应模型验证异常的处理

FastAPI也会验证响应模型,如果响应不符合模型定义,会抛出ValidationError

from pydantic import BaseModel, Field

class Item(BaseModel):
    id: int
    name: str
    price: float = Field(gt=0)

@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: int):
    # 模拟返回不符合模型的数据
    return {"id": item_id, "name": "Test", "price": -1}  # 价格为负数,会触发验证错误

5.4 验证错误信息的格式化

我们可以自定义验证错误信息的格式,使其更加友好:

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    formatted_errors = []
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"])
        message = error["msg"]
        error_type = error["type"]
        formatted_errors.append(f"{field}: {message} ({error_type})")
    
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "errors": formatted_errors
        }
    )

5.5 验证异常的常见问题与解决方案

问题1:验证错误信息不够友好

解决方案:自定义验证异常处理器,格式化错误信息

问题2:无法区分不同类型的验证错误

解决方案:根据错误类型进行分类处理

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    field_errors = {}
    for error in exc.errors():
        field = ".".join(str(loc) for loc in error["loc"])
        field_errors[field] = error["msg"]
    
    return JSONResponse(
        status_code=422,
        content={
            "detail": "Validation error",
            "field_errors": field_errors
        }
    )

问题3:嵌套模型的验证错误难以理解

解决方案:使用递归方式格式化嵌套模型的错误信息

def format_validation_errors(errors, prefix=""):
    formatted = {}
    for error in errors:
        loc = error["loc"]
        field = ".".join(str(l) for l in loc)
        if prefix:
            field = f"{prefix}.{field}"
        formatted[field] = error["msg"]
    return formatted

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

6. 数据库异常处理

6.1 数据库连接异常的捕获

数据库连接异常是常见的错误类型,我们可以使用try-except块来捕获并处理:

from sqlalchemy.exc import SQLAlchemyError

@app.get("/items")
async def get_items():
    try:
        items = db.query(Item).all()
        return items
    except SQLAlchemyError as e:
        # 记录异常
        logger.error(f"Database error: {e}", exc_info=True)
        raise HTTPException(
            status_code=500,
            detail="Database connection error"
        )

6.2 事务异常的处理策略

在数据库事务中,我们需要确保事务的原子性,即使发生异常也能正确回滚:

from sqlalchemy.orm import Session

@app.post("/transfer")
async def transfer(from_account: int, to_account: int, amount: float, db: Session = Depends(get_db)):
    try:
        # 开始事务
        # 检查余额
        from_user = db.query(User).filter(User.id == from_account).first()
        if from_user.balance < amount:
            raise InsufficientFundsException(from_user.balance, amount)
        
        # 执行转账
        from_user.balance -= amount
        to_user = db.query(User).filter(User.id == to_account).first()
        to_user.balance += amount
        
        # 提交事务
        db.commit()
        return {"message": "Transfer successful"}
    except Exception as e:
        # 回滚事务
        db.rollback()
        # 重新抛出异常
        raise

6.3 ORM框架异常的统一处理

我们可以为ORM框架的异常注册全局处理器:

from sqlalchemy.exc import SQLAlchemyError

@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
    # 记录异常
    logger.error(f"Database error: {exc}", exc_info=True)
    return JSONResponse(
        status_code=500,
        content={"detail": "Database operation failed"}
    )

6.4 数据库错误的重试机制

对于临时性的数据库错误,我们可以实现重试机制:

import asyncio
from sqlalchemy.exc import SQLAlchemyError

def retry_on_exception(max_retries: int = 3, delay: float = 0.5):
    def decorator(func):
        async def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    return await func(*args, **kwargs)
                except SQLAlchemyError as e:
                    retries += 1
                    if retries >= max_retries:
                        raise
                    logger.warning(f"Database error, retrying ({retries}/{max_retries}): {e}")
                    await asyncio.sleep(delay)
        return wrapper
    return decorator

@app.get("/items")
@retry_on_exception()
async def get_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

6.5 数据库异常的最佳实践

  1. 使用事务:确保数据库操作的原子性
  2. 捕获具体异常:根据不同的异常类型采取不同的处理策略
  3. 实现重试机制:对于临时性错误,实现自动重试
  4. 记录详细信息:记录异常的详细信息,便于调试和监控
  5. 提供友好的错误信息:向客户端返回友好的错误信息,不暴露内部实现细节

7. 异步异常处理

7.1 异步代码中的异常捕获

在异步代码中,我们需要使用try-except块来捕获异常:

@app.get("/async/items")
async def get_items():
    try:
        # 异步操作
        items = await async_db_query()
        return items
    except Exception as e:
        logger.error(f"Async operation failed: {e}", exc_info=True)
        raise HTTPException(
            status_code=500,
            detail="Async operation failed"
        )

7.2 异步任务异常的处理方法

对于后台异步任务,我们需要捕获并处理异常:

import asyncio
from fastapi import BackgroundTasks

async def send_email(email: str, message: str):
    try:
        # 发送邮件的异步操作
        await async_send_email(email, message)
    except Exception as e:
        logger.error(f"Failed to send email to {email}: {e}", exc_info=True)

@app.post("/send-email")
async def send_email_endpoint(email: str, message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, email, message)
    return {"message": "Email sent in background"}

7.3 WebSocket连接异常的处理

WebSocket连接异常需要特殊处理:

from fastapi import WebSocket, WebSocketDisconnect

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Message received: {data}")
    except WebSocketDisconnect:
        logger.info("WebSocket disconnected")
    except Exception as e:
        logger.error(f"WebSocket error: {e}", exc_info=True)
        await websocket.close(code=1011, reason="Internal server error")

7.4 异步上下文管理器中的异常

在异步上下文管理器中,我们需要确保即使发生异常也能正确清理资源:

class AsyncDatabaseConnection:
    async def __aenter__(self):
        # 建立数据库连接
        self.conn = await create_db_connection()
        return self.conn
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        # 关闭数据库连接
        await self.conn.close()
        # 如果发生异常,重新抛出
        if exc_type:
            return False

@app.get("/items")
async def get_items():
    async with AsyncDatabaseConnection() as conn:
        # 使用连接执行操作
        items = await conn.fetch_all("SELECT * FROM items")
        return items

7.5 异步异常的性能考虑

异步异常处理需要注意性能问题:

  1. 避免频繁抛出异常:异常处理会产生额外的性能开销
  2. 使用适当的异常粒度:不要为每个小错误都抛出异常
  3. 合理使用try-except块:只在必要的地方使用try-except块
  4. 异步异常的传播:确保异步异常能够正确传播到调用者

8. 异常处理与依赖注入

8.1 依赖注入中的异常处理

在依赖注入中,我们可以在依赖函数中抛出异常,这些异常会被FastAPI的异常处理器捕获:

from fastapi import Depends, HTTPException

async def get_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials"
        )
    return user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

8.2 异常在依赖链中的传递

当依赖链中有多个依赖时,异常会沿着依赖链向上传播:

async def get_db():
    db = SessionLocal()
    try:
        yield db
    except Exception as e:
        # 处理数据库异常
        logger.error(f"Database error in dependency: {e}", exc_info=True)
        raise
    finally:
        db.close()

async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
    user = verify_token(token, db)
    if not user:
        raise HTTPException(
            status_code=401,
            detail="Invalid authentication credentials"
        )
    return user

@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_user)):
    return current_user

8.3 依赖注入与全局异常处理的关系

依赖注入中的异常会被全局异常处理器捕获,因此我们可以为依赖中可能出现的异常注册全局处理器:

class DatabaseConnectionError(Exception):
    def __init__(self, message: str):
        self.message = message
        super().__init__(self.message)

@app.exception_handler(DatabaseConnectionError)
async def database_connection_error_handler(request: Request, exc: DatabaseConnectionError):
    return JSONResponse(
        status_code=503,
        content={"detail": f"Database connection error: {exc.message}"}
    )

async def get_db():
    try:
        db = SessionLocal()
        yield db
    except Exception as e:
        raise DatabaseConnectionError(str(e))
    finally:
        db.close()

8.4 依赖注入中异常的最佳实践

  1. 在依赖中抛出具体的异常:使用自定义异常类,便于识别和处理
  2. 在依赖中记录异常:在依赖函数中记录异常信息,便于调试
  3. 使用依赖链处理复杂逻辑:将复杂的异常处理逻辑分散到不同的依赖中
  4. 确保依赖的清理:使用try-finally块确保资源的正确清理

8.5 异常的优先级管理

当多个异常处理器都可以处理同一个异常时,FastAPI会按照以下顺序选择处理器:

  1. 最具体的异常类型的处理器
  2. 父类异常类型的处理器
  3. 全局异常处理器

我们可以利用这一特性来实现异常的优先级管理:

# 具体异常处理器
@app.exception_handler(DatabaseConnectionError)
async def database_connection_error_handler(request: Request, exc: DatabaseConnectionError):
    return JSONResponse(
        status_code=503,
        content={"detail": f"Database connection error: {exc.message}"}
    )

# 通用数据库异常处理器
@app.exception_handler(SQLAlchemyError)
async def sqlalchemy_exception_handler(request: Request, exc: SQLAlchemyError):
    return JSONResponse(
        status_code=500,
        content={"detail": "Database operation failed"}
    )

# 全局异常处理器
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )

9. 异常处理的源码与性能

9.1 FastAPI异常处理的核心机制

FastAPI的异常处理基于Starlette的异常处理机制,主要流程如下:

  1. 当请求处理过程中发生异常时,FastAPI会捕获该异常
  2. FastAPI会根据异常类型查找对应的异常处理器
  3. 如果找到匹配的处理器,会调用该处理器处理异常
  4. 如果没有找到匹配的处理器,会使用默认的异常处理器
  5. 异常处理器返回的响应会被发送给客户端

9.2 Starlette异常处理的底层实现

Starlette的异常处理核心代码位于starlette/exceptions.py文件中,主要包含以下类:

  • ExceptionMiddleware:中间件,负责捕获和处理异常
  • HTTPException:HTTP异常基类
  • WebSocketException:WebSocket异常基类

9.3 执行流程图

9.4 时序图

9.5 性能开销分析与优化

异常处理会产生一定的性能开销,主要包括:

  1. 异常捕获的开销:try-except块会增加代码的执行时间
  2. 异常处理器的执行开销:异常处理器的执行会增加响应时间
  3. 异常对象的创建开销:创建异常对象会分配内存

优化策略:

  1. 避免频繁抛出异常:只在真正的异常情况下抛出异常
  2. 使用快速路径:对于常见情况,使用条件判断而不是异常
  3. 优化异常处理器:保持异常处理器的简洁和高效
  4. 使用缓存:对于重复的异常处理逻辑,使用缓存

9.6 异常监控与告警机制

我们可以使用监控工具来监控异常的发生情况:

import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 记录异常
    logger.error(
        f"Unhandled exception: {exc}",
        exc_info=True,
        extra={
            "path": request.url.path,
            "method": request.method,
            "query_params": dict(request.query_params)
        }
    )
    
    # 发送告警
    send_alert(f"Unhandled exception: {exc}", {
        "path": request.url.path,
        "method": request.method
    })
    
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal server error"}
    )
Logo

腾讯云面向开发者汇聚海量精品云计算使用和开发经验,营造开放的云计算技术生态圈。

更多推荐