如何在Flask SQLAlchemy中监控SQL执行并获取超时SQL的栈信息？

问题根源 #

你的监听器未生效主要有两个核心原因：

• 未在before_cursor_execute事件中记录查询开始时间，导致context._query_start_time属性不存在
• 监听器注册时机滞后或代码结构问题，使得监听器未正确绑定到目标engine

完整解决方案 #

1. 正确配置双监听器 #

需要同时监听before_cursor_execute（记录查询起始时间）和after_cursor_execute（计算耗时、捕获慢查询），且监听器必须在engine创建后立即注册，确保所有SQL执行都能被捕获。

2. 可运行代码示例 #

import logging
import time
import traceback
from datetime import datetime
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker

# 配置日志格式，确保能清晰输出慢SQL及栈信息
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def test_sql():
    HOST = "127.0.0.1"
    PORT = "3306"
    DATABASE = "6023_xm"
    USERNAME = "root"
    PASSWORD = "root"

    # 创建数据库连接engine
    engine = create_engine(
        "mysql+pymysql://{}:{}@{}:{}/{}?charset=utf8mb4".format(USERNAME, PASSWORD, HOST, PORT, DATABASE)
    )

    # 注册前置监听器：记录查询开始时间
    @event.listens_for(engine, "before_cursor_execute")
    def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        context._query_start_time = time.time()

    # 注册后置监听器：计算耗时并捕获慢SQL
    @event.listens_for(engine, "after_cursor_execute")
    def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
        total_time = time.time() - context._query_start_time
        # 过滤执行时间超过3秒的SQL
        if total_time > 3:
            # 过滤掉SQLAlchemy内部栈帧，只保留业务代码调用链
            stack_frames = traceback.extract_stack()
            filtered_stack = [frame for frame in stack_frames if "sqlalchemy" not in frame.filename]
            stack_str = "".join(traceback.format_list(filtered_stack))
            
            # 输出告警信息
            logger.warning(f"""
慢SQL告警：
执行时长：{total_time:.2f}秒
SQL语句：{statement}
参数：{parameters}
业务调用栈：
{stack_str}
            """)

    # 创建session并绑定engine
    Session = sessionmaker(bind=engine)
    session = Session()

    # 确保业务代码使用当前session（避免使用未注册监听器的连接）
    UpdateShuJuKu.qy_sjk_update(session)

if __name__ == "__main__":
    test_sql()

关键注意事项 #

• 监听器注册时机：必须在engine创建后立即注册，否则后续创建的连接不会触发监听器
• 调用栈过滤：通过过滤sqlalchemy相关栈帧，避免日志被框架内部代码淹没，聚焦业务调用路径
• Session传递：业务代码qy_sjk_update必须使用当前创建的session，否则可能使用未绑定监听器的其他连接
• 日志级别：使用warning级别输出慢SQL，便于在日志系统中快速识别告警

额外优化建议 #

• 可将监听器逻辑抽离为独立函数，避免在业务函数内嵌套，提升代码可维护性
• 若在Flask全局应用中使用，可在应用初始化阶段注册监听器，确保所有请求的SQL执行都被监控

内容的提问来源于stack exchange，提问作者li linfei