如何借助sentry_sdk.capture_event实现Python定时脚本任务性能追踪？

1. 初始化Sentry SDK #

先确保安装sentry-sdk，初始化时开启性能追踪采样：

import sentry_sdk
from sentry_sdk import capture_event
from sentry_sdk.utils import utcnow, get_stacktrace

sentry_sdk.init(
    dsn="你的Sentry项目DSN",
    traces_sample_rate=1.0,  # 开发环境全量采样，生产环境可按需调整比例
)

2. 封装任务性能追踪函数 #

针对批量相似任务，写一个通用包裹函数，自动生成并发送性能事件：

def track_task(task_name, task_func, *args, **kwargs):
    # 构造事务事件基础信息
    base_event = {
        "type": "transaction",
        "transaction": task_name,
        "start_timestamp": utcnow(),
        "contexts": {
            "trace": {
                "trace_id": sentry_sdk.Hub.current.scope.transaction.trace_id,
                "span_id": sentry_sdk.Hub.current.scope.transaction.span_id,
                "op": "custom_task",  # 自定义操作标识，用于Sentry后台筛选分类
                "status": "ok"
            }
        }
    }

    try:
        # 执行目标任务
        result = task_func(*args, **kwargs)
        
        # 生成任务完成事件，计算耗时
        end_time = utcnow()
        duration = (end_time - base_event["start_timestamp"]).total_seconds()
        
        finish_event = base_event.copy()
        finish_event["timestamp"] = end_time
        finish_event["duration"] = duration
        finish_event["status"] = "ok"
        
        capture_event(finish_event)
        return result
    except Exception as e:
        # 任务失败时生成异常关联事件
        end_time = utcnow()
        duration = (end_time - base_event["start_timestamp"]).total_seconds()
        
        error_event = base_event.copy()
        error_event["timestamp"] = end_time
        error_event["duration"] = duration
        error_event["status"] = "internal_error"
        error_event["exception"] = {
            "values": [{
                "type": type(e).__name__,
                "value": str(e),
                "stacktrace": get_stacktrace()
            }]
        }
        
        capture_event(error_event)
        raise  # 保留原有异常抛出逻辑，不中断脚本流程

3. 批量执行任务并追踪 #

直接用track_task包裹你的任务逻辑即可：

# 示例任务函数，替换为你的实际业务代码
def process_data(data_id):
    print(f"完成数据处理: {data_id}")

# 遍历执行所有任务
for data_id in range(1, 5):
    track_task(f"数据处理任务_{data_id}", process_data, data_id)

关键注意点 #

• 采样率配置：traces_sample_rate必须大于0，否则性能事件不会被Sentry接收
• 自定义标识优化：可修改contexts.trace.op为业务相关值（如data_processing、third_party_api_call），方便在Sentry性能页面快速筛选任务类型
• 异常关联：函数自动捕获任务异常并关联至性能事件，在Sentry中可直接从性能页面跳转查看错误栈信息

内容的提问来源于stack exchange，提问作者Ryabchenko Alexander