如何为FastAPI路由编写Sentry错误与追踪采样率配置装饰器?
实现FastAPI路由级别的Sentry错误与追踪采样率控制
方案可行性
这个方案完全可行,sentry_sdk支持通过**请求作用域(Scope)**动态修改单个请求的采样规则,无需全局配置,刚好适配路由级别的定制需求。
核心实现思路
- 追踪采样率:利用
sentry_sdk.configure_scope()获取当前请求的作用域,直接设置traces_sample_rate,该设置仅对当前请求生效。 - 错误采样率:同样通过请求作用域设置错误采样参数,结合异常捕获逻辑,确保异常时的采样规则符合路由配置。
完整代码实现
import sentry_sdk from functools import wraps from typing import Callable, Awaitable def sentry_trace_rate(traces_sample_rate: float = 0.0) -> Callable: """为指定路由设置追踪采样率""" def decorator(func: Callable) -> Awaitable: @wraps(func) async def wrapper(*args, **kwargs): # 获取当前请求的Sentry作用域,修改追踪采样率 with sentry_sdk.configure_scope() as scope: scope.set_traces_sample_rate(traces_sample_rate) return await func(*args, **kwargs) return wrapper return decorator def sentry_error_rate(errors_sample_rate: float = 0.0) -> Callable: """为指定路由设置错误采样率""" def decorator(func: Callable) -> Awaitable: @wraps(func) async def wrapper(*args, **kwargs): # 设置当前请求的错误采样率 with sentry_sdk.configure_scope() as scope: # 错误采样通过scope的sample_rate控制 scope.set_sample_rate(errors_sample_rate) try: return await func(*args, **kwargs) except Exception as e: # 确保异常被Sentry捕获时使用当前路由的采样率 sentry_sdk.capture_exception(e) raise # 重新抛出异常,不影响原有业务逻辑 return wrapper return decorator
使用示例
在FastAPI路由上直接应用装饰器即可:
from fastapi import FastAPI app = FastAPI() # 高频率路由设置低采样率 @app.get("/health") @sentry_trace_rate(traces_sample_rate=0.01) @sentry_error_rate(errors_sample_rate=0.1) async def health_check(): return {"status": "ok"} # 核心业务路由设置全量采样 @app.post("/api/order") @sentry_trace_rate(traces_sample_rate=1.0) @sentry_error_rate(errors_sample_rate=1.0) async def create_order(): # 业务逻辑 return {"order_id": "12345"}
注意事项
- 确保
sentry_sdk已完成基础初始化,例如:sentry_sdk.init( dsn="YOUR_SENTRY_DSN", # 全局默认采样率,路由装饰器会覆盖该值 traces_sample_rate=0.5, sample_rate=0.5 ) - 采样率参数需在
0.0到1.0之间,0.0表示完全不采样,1.0表示全量采样。 - 装饰器支持异步路由(FastAPI默认推荐异步写法),如果是同步路由,只需将
async def wrapper改为def wrapper,并去掉await关键字。
内容的提问来源于stack exchange,提问作者Yohann L.
相关产品推荐
相关产品推荐

