FastAPI异常处理全面指南
在FastAPI中,我们可以通过继承Exception# 自定义异常类# 继承自HTTPException).dict()我们可以通过继承# 使用自定义HTTP异常。
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 异常处理的基本原则
- 早发现早处理:在错误发生的最早阶段捕获并处理异常
- 统一处理:使用全局异常处理器统一处理相同类型的异常
- 提供清晰的错误信息:确保错误信息对客户端友好且有帮助
- 记录异常:记录异常信息以便于调试和监控
- 避免暴露敏感信息:不在错误响应中包含敏感信息
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:要处理的异常类型- 处理函数:接收
request和exc两个参数,返回响应
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:要处理的异常类型,可以是内置异常或自定义异常- 处理函数:一个异步函数,接收
request和exc两个参数,返回响应对象
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会按照异常处理器的注册顺序来匹配异常类型,具体规则如下:
- 首先尝试匹配最具体的异常类型
- 如果没有找到匹配的处理器,会尝试匹配父类异常类型
- 最后会使用默认的异常处理器
因此,我们应该先注册具体异常的处理器,再注册通用异常的处理器:
# 正确的注册顺序
@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异常的最佳实践
- 使用具体的状态码:根据错误类型选择合适的HTTP状态码
- 提供清晰的错误信息:确保错误信息对客户端友好且有帮助
- 使用自定义HTTP异常:为常见错误创建自定义HTTP异常类,提高代码可读性
- 记录异常信息:在抛出异常前记录相关信息,便于调试和监控
- 避免暴露敏感信息:不在错误响应中包含敏感信息
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 数据库异常的最佳实践
- 使用事务:确保数据库操作的原子性
- 捕获具体异常:根据不同的异常类型采取不同的处理策略
- 实现重试机制:对于临时性错误,实现自动重试
- 记录详细信息:记录异常的详细信息,便于调试和监控
- 提供友好的错误信息:向客户端返回友好的错误信息,不暴露内部实现细节
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 异步异常的性能考虑
异步异常处理需要注意性能问题:
- 避免频繁抛出异常:异常处理会产生额外的性能开销
- 使用适当的异常粒度:不要为每个小错误都抛出异常
- 合理使用try-except块:只在必要的地方使用try-except块
- 异步异常的传播:确保异步异常能够正确传播到调用者
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 依赖注入中异常的最佳实践
- 在依赖中抛出具体的异常:使用自定义异常类,便于识别和处理
- 在依赖中记录异常:在依赖函数中记录异常信息,便于调试
- 使用依赖链处理复杂逻辑:将复杂的异常处理逻辑分散到不同的依赖中
- 确保依赖的清理:使用try-finally块确保资源的正确清理
8.5 异常的优先级管理
当多个异常处理器都可以处理同一个异常时,FastAPI会按照以下顺序选择处理器:
- 最具体的异常类型的处理器
- 父类异常类型的处理器
- 全局异常处理器
我们可以利用这一特性来实现异常的优先级管理:
# 具体异常处理器
@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的异常处理机制,主要流程如下:
- 当请求处理过程中发生异常时,FastAPI会捕获该异常
- FastAPI会根据异常类型查找对应的异常处理器
- 如果找到匹配的处理器,会调用该处理器处理异常
- 如果没有找到匹配的处理器,会使用默认的异常处理器
- 异常处理器返回的响应会被发送给客户端
9.2 Starlette异常处理的底层实现
Starlette的异常处理核心代码位于starlette/exceptions.py文件中,主要包含以下类:
ExceptionMiddleware:中间件,负责捕获和处理异常HTTPException:HTTP异常基类WebSocketException:WebSocket异常基类
9.3 执行流程图

9.4 时序图

9.5 性能开销分析与优化
异常处理会产生一定的性能开销,主要包括:
- 异常捕获的开销:try-except块会增加代码的执行时间
- 异常处理器的执行开销:异常处理器的执行会增加响应时间
- 异常对象的创建开销:创建异常对象会分配内存
优化策略:
- 避免频繁抛出异常:只在真正的异常情况下抛出异常
- 使用快速路径:对于常见情况,使用条件判断而不是异常
- 优化异常处理器:保持异常处理器的简洁和高效
- 使用缓存:对于重复的异常处理逻辑,使用缓存
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"}
)更多推荐
所有评论(0)